Confluent-kafka-dotnet: How to use a custom partitioner ?

[mgrail]

Hello all,

Thank you for your work.

I need to use a partitioner but I couldn't find any way to set it.
In the advanced producer sample, there is a comment line 42 where @mhowlett use an older way to create a partitioner (with TopicConfig class) but it doesn't exist anymore ?

Thank you very much

[mhowlett]

unfortunately this is not implemented yet @mgrail. it's commonly requested, but I don't have a timeframe yet. do you want this so you can provide a murmur2 partitioner compatible with the java client? or a custom strategy?

[mgrail]

Hi,

It was only to round robin on all partitions based on a message key (int).
I'll use ProduceAsync to give a partition instead.

Thank you

[edenhill]

We're looking to provide the builtin partitioners as a simple configuration property in a future release.

[dipanshah]

Hi @mhowlett \ @edenhill ,

I have run into an issue which seems to be related to this.

We are using confluent-kafka-dotnet 0.9.5. The topic has 50 partitions but when I produce a message with the key as NULL, all the messages are going into the same partition.

I looked at Apache's documentation for Kafka and it seems that org.apache.kafka.clients.producer.Partitioner is used to set the Random or Default partitioner. But I am not able to find anything related to that in confluent-kafka-dotnet.

Can you please help me with this as to how I could set up Random or Round Robin partitioning here?

Thanks.

[treziac]

Can you post a snippet of your producer code? You should use a null key and not use the Produce overrides which take partition.

By default, a consistent random partitioner is used (you may find more about this in librdkafka - basically, it's juste a random between 0 and partitionCount)

Also be sure your topic has more than one insync partition

[dipanshah]

Hello @treziac,

I am not using any overrides for the partitioner. Still here's the code:

var Config = new Dictionary<string, object> {
    { "group.id", "topic1_group" },
    { "bootstrap.servers", "localhost:9092" },
    { "default.topic.config",new Dictionary<string, object> {
                               { "acks", 1}
                            }
    }
};

var producer = new Producer<string, string>(Config, new StringSerializer(Encoding.UTF8), new StringSerializer(Encoding.UTF8));
var producerResult = producer.ProduceAsync("topic1", null, "Message1");

Thanks for helping.

[mtrtm]

I'd love to know when this work is scheduled for - we are evaluating this as a possible production client, but obviously cannot mix this with the Java client when the partitioning is different. Are you looking for any help on it? I did not see that you are when looking in the 'help wanted' section...

[mhowlett]

I notice that librdkafka recently added a murmur2 partitioner (same as java) https://github.com/edenhill/librdkafka/pull/1468

it should be possible to make use of this via dotnet but it would require modifications to the dotnet library. i was hoping this functionality would be made available via a new librdkafka config property (which would make it available in the language bindings without any modification). any plans for this @edenhill ?

[edenhill]

Yes, the plan is to initially expose the builtin partitioners as string config properties.
Going forward we might want to make it possible to register new custom partitioners with the configuration interface so they too can be configured this way.

[edenhill]

The next version of librdkafka (the underlying Kafka client) will expose the builtin partitioners as configuration properties, allowing you to change to an alternative builtin partitioner, such as the Java compatible murmur2_random partitioner.

Search for 'partitioner' here: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

Custom partitioners are not yet exposed in the .NET client though.

[sfrooster]

@mhowlett any estimate as to when this might find its way in? We first started asking about this in early May 2017 https://github.com/confluentinc/confluent-kafka-dotnet/issues/165 , followed-up again in early August https://github.com/confluentinc/confluent-kafka-dotnet/issues/274, and now we're having customer issues because of this one issue. Really looking forward to addressing this.

[mhowlett]

@sfrooster - do you just need a partitioner matching the java client (murmur2_random), or custom partitioner functionality? the option to use the murmur2_random partitioner will be in the next release version.

[sfrooster]

@mhowlett murmur2_random matching the java client would suffice. Are we talking about 0.11.4 then?

[mhowlett]

the next release that aligns with a librdkafka release. we don't have a target date yet, but it shouldn't be too long. I may do a .net release before that to get SR/Avro integration out.

[sfrooster]

@mhowlett We've been getting along fine without this, but would like to switch using this, once available. Any update?

[mhowlett]

iirc this made it in the 0.11.4 release, but there was a bug (didn't completely match java). there is now a fix for that in librdkfaka master and the next release (0.11.5 and 1.0 pre-release) will have a partitioner exactly matching the default java one. I think the timeline for that is about a month.

Also, custom partitioners are likely to make it into the 1.0-experimental branch soon.

[sfrooster]

that's good to know - is there a roadmap or some such other that shows when 0.11.5 will be available in nuget?

[sfrooster]

We have already worked around this, for what's it worth, but yes - all we're currently interested in is a partitioner that matches the default java murmur2 partitioner. At this point, the benefit to us is being able to remove a background task whose job it is to occasionally refresh metadata/topic-partition-counts (so we can mod the murmur2 hash appropriately). Maybe we've outthought ourselves and that's not necessary - dunno. But if it's implemented "natively" then we can definitely strip that code out.

Last I recall 0.11.5 was scheduled for 6/26 (do I have that right?) - is that still the target?

Thanks!

[mhowlett]

0.11.5 shouldn't be far - a librdkafka rc has or is about to happen. Version experimental-4 already links against librdkafka 0.11.5-pre5, which contains this fix.

[sfrooster]

Just checking in on this and any possible date for 0.11.5?

[mhowlett]

I don't have an exact date, though note that librdkafka is up to RC2 and the dotnet release will coincide.

[mhowlett]

0.11.5 is out. also, we've worked out what the API will look like for user-defined partitioners (harder than implementing the feature!). that's coming too, promise!

[mhowlett]

custom partitioner enhancement is being tracked in #587