Producer Refactor Mk II #8

mhowlett · 2016-11-21T17:59:56Z

create serializing producer: new Producer<TKey, TValue>(...)
create untyped producer: new Producer(...)
ability to wrap untyped producer to add serializers if desired.
ArraySegment capability to SafeTopicHandle and up (marshaling impl. similar to open PR on rdkafka-dotnet).

mhowlett · 2016-11-21T18:05:22Z

src/Confluent.Kafka/Impl/SafeTopicHandle.cs

+                return (long) LibRdKafka.produce(
+                        handle,
+                        partition,
+                        (IntPtr) (MsgFlags.MSG_F_COPY | (blockIfQueueFull ? MsgFlags.MSG_F_BLOCK : 0)),


how long does librdkafka need the memory for? rather than copying it, we could pin it until after the delivery report comes back. this may or may not be more performant.

Until the delivery report callback returns.

Re .._MSG_F_BLOCK:
since the message queue threshold limit also includes delivery reports, some other thread of the application will need to call poll() for a blocking produce() to ever unblock.

Re: .._MSG_F_BLOCK - there is a thread in Handle which is devoted to calling poll. If the Task based ProduceAsync methods are used, I believe the continuations are always on a different thread, so there will never be a problem. If the callback ProduceAsync methods are used, and the callbacks produce messages, then there is potentially a problem though.

Re memcpy vs pinning, I'm currently thinking I'll leave as is for this version and possibly investigate in a future version.

mhowlett · 2016-11-21T18:06:14Z

src/Confluent.Kafka/Impl/SafeTopicHandle.cs

+
+            if (val != null)
+            {
+                gchValue = GCHandle.Alloc(val, GCHandleType.Pinned);


I expect marshaling of byte[] does this guff behind the scenes.

mhowlett · 2016-11-21T18:07:29Z

src/Confluent.Kafka/Impl/SafeTopicHandle.cs

+            if (val != null)
+            {
+                gchValue = GCHandle.Alloc(val, GCHandleType.Pinned);
+                pValue = Marshal.UnsafeAddrOfPinnedArrayElement(val, valOffset);


the safety of this is enforced higher up by use of ArraySegment which performs run time bounds checking on parameters.

this should arguably (would be better) enforced here. But then we get run time bounds checking twice. yay! maybe we do want (byte[], int, int) in the API after all instead of ArraySegment.

mhowlett · 2016-11-21T18:10:52Z

src/Confluent.Kafka/Producer.cs

+        public Task<DeliveryReport> ProduceAsync(string topic, byte[] key, byte[] val, int? partition = null, bool blockIfQueueFull = true)
+            => getKafkaTopic(topic).Produce(val, 0, val.Length, key, 0, key.Length, partition ?? RD_KAFKA_PARTITION_UA, blockIfQueueFull);
+
+        public Task<DeliveryReport> ProduceAsync(string topic, ArraySegment<byte> key, ArraySegment<byte> val, int? partition = null, bool blockIfQueueFull = true)


I've used ArraySegment here rather than (byte[], int, int) parameters explicitly. This is at odds with many methods in the .NET Framework (eg. BitConverter.ToString). I don't know if there is a good reason for this or whether it's a function of the relative age of ArraySegment struct compared to these other methods. One benefit is ArraySegment provides runtime bounds checking (we could do this explicitly of course though). Note that ArraySegment is a struct (stack allocated), so no GC overhead.

mhowlett · 2016-11-21T18:22:21Z

examples/AdvancedProducer/Program.cs

            var config = new Dictionary<string, object> { { "bootstrap.servers", brokerList } };

-            using (var producer = new Producer<string, string>(config))
+            using (var producer = new Producer<string, string>(config, new StringSerializer(), new StringSerializer()))


@ewencp argues that these should be explicit, as most of the time defaults won't be obvious so most of the time if they're left off this has a good chance of being a user error. I agree, though It's kind of a shame.

mhowlett · 2016-11-21T18:33:54Z

examples/Wrapped/Program.cs

+                var sProducer2 = producer.Wrap<Null, int>(new NullSerializer(), new IntSerializer());
+
+                // write (string, string) data to topic "first-topic", statically type checked.
+                sProducer1.ProduceAsync("first-topic", "my-key-value", "my-value");


oh, i forgot to wait on these tasks.

As mentioned elsewhere, ideally this could just be a Flush() on the unwrapped producer.

For that matter, this does raise the question of how "aggregate" operations behave for the wrapper producers. i.e. would Flush() in any way isolate itself to messages for the single format (presumably no, given the way this is implemented)? How about things like Dispose?

I don't think we want Flush (though I'm not 100% sure i'm seeing the world correctly here). I think we just want to wait on the tasks, or a collection of tasks.

If we have a flush method, it doesn't make sense to put it on ISerializingProducer. It'd be on the concrete Producer and Producer<TKey, TValue> only.

The dispose method of Producer effectively flushes.

I think it makes sense to have en explicit Flush() method, and not to do an implicit Flush() when disposing, to allow applications to exit quickly without waiting for message transmission (which may block for a long time)

Good point about not wanting to flush in Dispose... shutdown is not the only consideration - a using statement will typically be used to wrap a producer and this is equivalent to try / finally. We probably don't want messages being flushed before an exception gets handled.

I guess Flush is a good thing to have in addition to the Tasks, as some people will probably just want to fire and forget and ignore the Tasks.

On the other hand not flushing in .Dispose makes the API more error prone as users will often be in a situation where there may be messages in flight when they want to exit because all the calls are async. So, calling Flush will usually be an appropriate thing to do right at the end of the Producer using block.

Also, if an exception makes it outside the Producer scope, there is no reference to the producer, so no option to Flush.

Thoughts on having a property .FlushOnDispose, which by default is true?

None of the other clients does an implicit flush on dispose, so I dont think we should alter that behaviour in this client.

Also note that flush, and thus dispose, might block for up to message.timeout.ms which defaults to 5 minutes.

Yeah I agree.

I think I'm only still thinking about flush on dispose because that is the existing behavior of rdkafka-dotnet (and there are some positives to the idea).

But now I'm seeing it from a different point of view - it's not normal to wait a long time on dispose, so it's a counter intuitive thing to do.

I'll work out something similar to the other clients.

as indication that it's counter intuitive to flush in dispose, see my first comment in this thread "oh, i forgot to wait on these tasks." - this was before I started thinking about what was happening in the dispose method and before people started suggesting a flush method. my intuition then was dispose was not going to wait on anything.

ewencp · 2016-11-21T20:36:31Z

examples/Benchmark/Program.cs

-        public static void Produce(string broker, string topicName, long numMessages)
-        {
-            var deliveryHandler = new DeliveryHandler();
+                public static void WaitForAllDeliveryReports()


Are we just lacking Flush() right now? That's how I'd normally expect to wait for all sends to complete.

For this benchmark, I'm using the non-Task ProduceAsync method. Probably a premature optimization to even have this option. Maybe we want to get rid of this. One property of this is all the callbacks happen on the same thread.

Anyway, with Tasks, you can do WaitAll on a collection of Tasks, and that is probably the idiomatic way to 'flush'.

Also, if order is guaranteed, you could also just wait on the last Task. I think if messages are produced to different partitions though, order is not guaranteed?

If we include the non task based methods, you're right, Flush is probably arguably necessary.

currently thinking they should be removed.

I just did another benchmark test using the Task produce method / WaitAll. It's substantially terser, and there is no noticeable change in perf. But I can't easily profile memory usage - The Task way makes an addition 5M objects.

The callback methods are here because they're in rdkafka-dotnet and I wanted to take them out only after careful consideration.

I'm taking them out. I can't think of any strong reason to have them and if it turns out there is one, it'll be much easier to change the API to add them back in than take them out.

ewencp · 2016-11-21T20:44:10Z

examples/Benchmark/Program.cs

+
+                        byte cnt = 0;
+                        var val = new byte[100].Select(a => ++cnt).ToArray();
+                        var key = new byte[0];


Is this required to be byte[0] rather than null? I'd normally expect null. I think they end up having the same overhead assuming compression is off, but strictly speaking they are different since null gets encoded as a length of -1 and a zero length array is encoded as 0 followed by 0 bytes. Just want to make sure we're not losing the ability to encode null in this patch.

Are you sure about the len == -1 encoding of null? cc @edenhill - quickly looking in rdkafka_msg.c it looks like it gets set to 0 to me if the data is null.

rdkafka-dotnet handles null ok and sets the length to 0.

and whoops, i'd previously noticed this and forgot to make null work.

librdkafka treats key=NULL as Kafka null key, while key!=NULL and size=0 as an empty key, which are not the same thing.
We should allow for the same semantics in this client.

Value is identical.

ewencp · 2016-11-21T20:51:08Z

examples/Wrapped/Program.cs

+                var sProducer2 = producer.Wrap<Null, int>(new NullSerializer(), new IntSerializer());
+
+                // write (string, string) data to topic "first-topic", statically type checked.
+                sProducer1.ProduceAsync("first-topic", "my-key-value", "my-value");


As mentioned elsewhere, ideally this could just be a Flush() on the unwrapped producer.

For that matter, this does raise the question of how "aggregate" operations behave for the wrapper producers. i.e. would Flush() in any way isolate itself to messages for the single format (presumably no, given the way this is implemented)? How about things like Dispose?

ewencp · 2016-11-21T21:08:25Z

src/Confluent.Kafka/DeliveryReport.cs

@@ -0,0 +1,8 @@
+namespace Confluent.Kafka
+{
+    public struct DeliveryReport


The equivalent in the Python client fills in more information -- the callbacks accept (err, msg) parameters, where the latter is http://docs.confluent.io/3.1.0/clients/confluent-kafka-python/index.html#confluent_kafka.Message Same deal with Go where the DR channel gets one of these: http://docs.confluent.io/3.1.0/clients/confluent-kafka-go/index.html#Message Is this being kept more minimal intentionally?

This is unchanged from rdkafka-dotnet.
If we include the message in the delivery report (given the precedent, we should), we're going to need to think about generics here too. I propose doing this in the next PR which is going to be a refactor of the Consumer.

A delivery report will need at least:

topic

partition

offset

error

msg opaque (or bound variable through other means)

Other stuff that might be useful:

value-object

key-object

value

key

timestamp

future fields that the community makes up, e.g. headers

Wrapping this in a Message type is consistent with other clients.

ewencp · 2016-11-21T21:16:49Z

src/Confluent.Kafka/Null.cs

 {
    public sealed class Null
    {
+        public static Null Instance = new Null();


Why do we need this? The point of this is that you can't instantiate it, right? It's not Singleton, it's Null.

It's 'cause I screwed up thinking about the deserializer. it's unnecessary.

ewencp · 2016-11-21T21:22:13Z

src/Confluent.Kafka/Producer.cs

+    }
+
+
+    public class Producer<TKey, TValue> : ISerializingProducer<TKey, TValue>, IDisposable


I don't think this should implement IDisposable. Supposedly you should only implement IDisposable if you directly handle unmanaged resources. In this case you are just using an IDisposable.

This definitely needs to implement IDisposable otherwise producer's resources can't be deterministically cleaned up.

Ack, I think it's easy to lose track of where the disposable pattern vs SafeHandle needs to be used. Seems we have one layer of SafeHandles around the underlying C resources and then use IDisposable everywhere else to allow proactive cleanup.

I think part of the confusion comes from overloading what IDisposable means. The docs for IDisposable even say

Implement IDisposable only if you are using unmanaged resources directly. If your app simply uses an object that implements IDisposable, don't provide an IDisposable implementation.

Unfortunately it seems people also use this to also be the equivalent of Closeable in Java (which doesn't need to imply anything about whether something will be garbage collected even if you don't call close().

Yes. Note there is still some cleaning up to do I think when I get to reviewing Handle. Also wanted to note that inheritance makes this more difficult to think through properly.

ewencp · 2016-11-21T21:24:31Z

src/Confluent.Kafka/Producer.cs

+
+        public void Dispose()
+        {
+            producer.Dispose();


I'm pretty sure this implementation is incorrect anyway. If I call wrap() twice and then Dispose() on the first wrapper, I'll dispose all the underlying resources and the other ISerializingProducer will break. Docs seem to indicate a SafeHandle or implementing Finalize on the wrapped class is the way to fix this.

You can't call Dispose on a wrapped object (Wrap returns ISerializingProducer, and concretely the internal SerializingProducer).

If you call Dispose on producer that is wrapped, the SerializingProducer will no longer work and will throw some sort of an exception if it's tried to be used. I could be more explicit about detecting this and throw a more explicit exception.

No one's going to do that in practice, I don't see it as a problem with the concept.

ewencp · 2016-11-21T21:31:51Z

src/Confluent.Kafka/Producer.cs

+    }
+
+
+    public class Producer<TKey, TValue> : ISerializingProducer<TKey, TValue>, IDisposable


How do docs work for classes with the same name but different # of generic parameters? Are we going to have to constantly maintain duplicate docstrings (as I assume we are going to fill these all in to get automated generation of docs)?

I'm not 100% sure, but guess it's going to mean duplication.
Yes, I'll fill all these out, but want to get the API right first.

ewencp · 2016-11-21T21:34:16Z

src/Confluent.Kafka/Serialization/NullDeserializer.cs

+    {
+        public Null Deserialize(byte[] data)
+        {
+            return Null.Instance;


This is kind of weird. I can put a null into a serializer and get an object back out of the deserializer...

yes, you're right, this is completely weird and unnecessary. I didn't think too hard about this yet as deserializers aren't used yet.

ewencp · 2016-11-21T21:47:29Z

src/Confluent.Kafka/Producer.cs

+
+    public class Producer<TKey, TValue> : ISerializingProducer<TKey, TValue>, IDisposable
+    {
+        private Producer producer;


Both should be readonly as they are always set in the constructor and should never change.

edenhill · 2016-11-22T13:26:09Z

examples/Benchmark/Program.cs

+                    { "queue.buffering.max.messages", 500000 },
+                    { "message.send.max.retries", 3 },
+                    { "retry.backoff.ms", 500 },
+                    { "queued.min.messages", 1000000 },


this is a consumer property, and so is session.timeout.ms.

oh, your example probably tests consumer as well. removed.

edenhill · 2016-11-22T13:26:29Z

examples/Benchmark/Program.cs

+                var config = new Dictionary<string, object>
+                {
+                    { "bootstrap.servers", broker },
+                    { "queue.buffering.max.messages", 500000 },


Is there a reason for setting these properties?

"mirrors the librdkafka performance test example.". Was trying to get something directly comparable to your numbers.

edenhill · 2016-11-22T13:27:58Z

examples/Benchmark/Program.cs

+
+                        byte cnt = 0;
+                        var val = new byte[100].Select(a => ++cnt).ToArray();
+                        var key = new byte[0];


librdkafka treats key=NULL as Kafka null key, while key!=NULL and size=0 as an empty key, which are not the same thing.
We should allow for the same semantics in this client.

Value is identical.

edenhill · 2016-11-22T13:29:27Z

examples/SimpleProducer/Program.cs

            var config = new Dictionary<string, object> { { "bootstrap.servers", brokerList } };

-            using (var producer = new Producer<Null, string>(config))
+            using (var producer = new Producer<Null, string>(config, new NullSerializer(), new StringSerializer()))


what's a NullSerializer and how is it different from not setting a serializer?

The choice is to either have a NullSerializer class or an explicit check whether the serializer is null in the ProduceAsync method. I'm sort of on the fence here, erring on the side of having NullSerializer.

Or simply defaulting to NullSerializer if null is passed as serializer in the constructor.
Not sure if this matters though. @ewencp ?

edenhill · 2016-11-22T13:31:01Z

examples/SimpleProducer/Program.cs

-                // TODO: There should be no need to specify a serializer for common types like string - I think it should default to the UTF8 serializer.
-                producer.ValueSerializer = (ISerializer<string>)new Confluent.Kafka.Serialization.Utf8StringSerializer();
-
                Console.WriteLine($"{producer.Name} producing on {topicName}. q to exit.");


Ctrl-c baby, not "q".

I've decided I disagree. The purpose of these examples is to demonstrate usage of the client in a straightforward, easy to understand way. Turns out that detecting Ctrl-C is a bit convoluted, in fact there are as many lines dedicated to doing this properly as demonstrating the producer. None of the code is rocket science of course, so i'm sort of indifferent, but on balance, I think using q to exit is better.

This might seem like a tiny nitpick thing, but there are two proper reasons:

examples should be correct, even if for unrelated stuff, people will base their own code on this.

out-of-band cancellation shows an interesting problem: how do I break out of the consume loop. If we can't show people how to do that in an effective and correct manner they will get it wrong and that will bite us back.

Is AppDomain.ProcessExit not a workable solution that will be relatively small?

regarding #1 - i wouldn't call using 'q' to exit 'incorrect' as such.
regarding #2 - you've convinced me it's useful. People won't be using Console.ReadLine, but many will be making console apps, and the CancelKeyPress handler is the way to detect Ctrl-C.

how about i leave it in the advanced producer example and keep it out of the simple producer example - keep that as dead simple as possible - i want the first example people look at to be inviting and not scary in any way.

@ewencp AppDomain is changed a lot or doesn't exist in .net core. http://www.michael-whelan.net/replacing-appdomain-in-dotnet-core/

The new assembly unload event mentioned in the above article is not effective at catching Ctrl-C. Examples I see around the web use Console.CancelKeyPress. I'm not certain there is not a better way, but it seems likely CancelKeyPress is good.

Another reason not to include this: there are higher priorities than figuring this out.

edenhill · 2016-11-22T13:46:44Z

src/Confluent.Kafka/Producer.cs

+            // TODO: specify serializers via config file as well? In which case, default keySerializer and valueSerizlizer params to null.
+
+            if (KeySerializer == null)
+            {


Do we really require a KeySerializer? Can't we allow this to be null instead of having the phony NullSerializer thingie?

I added that option of using null rather than NullSerializer. Can remove NullSerializer if that is the consensus. See other comment.

edenhill · 2016-11-22T13:49:06Z

src/Confluent.Kafka/Serialization/StringSerializer.cs

+    ///     <paramref name="val" /> cannot be null.
+    ///     TODO: well it shouldn't be other there is ambiguity on deserialization. check this.
+    /// </remarks>
+    public class StringSerializer : ISerializer<string>


Is it safe to assume that people will know this means UTF-8? Maybe being explicit about it is better

i'm on the fence on this. removed.

edenhill · 2016-11-22T13:50:38Z

src/Confluent.Kafka/Serialization/IntSerializer.cs

+
+namespace Confluent.Kafka.Serialization
+{
+    public class IntSerializer : ISerializer<int>


Document what serialization this is in practise.
Big endian? varint?

depends on architecture. added remark (will do propper docs later).

Huhm, you sure? A serializer shouldn't depend on the architecture, that's what makes the serialized format portable.

yes. maybe i need to go home for a nap.

edenhill · 2016-11-22T13:51:45Z

src/Confluent.Kafka/Topic.cs

-        }
-
-        public Task<DeliveryReport> Produce(byte[] val, int valLength, byte[] key = null, int keyCount = 0, Int32 partition = RD_KAFKA_PARTITION_UA, bool blockIfQueueFull = true)
+        public Task<DeliveryReport> Produce(byte[] val, int valOffset, int valLength, byte[] key = null, int keyOffset = 0, int keyLength = 0, Int32 partition = RD_KAFKA_PARTITION_UA, bool blockIfQueueFull = true)


This valOffset API is funky, aren't there slices or similar in .NET?

I'll leave this for now. will get rid of it when I refactor out the topic method.

edenhill · 2016-11-22T13:52:20Z

src/Confluent.Kafka/Topic.cs

+                {
+                    throw RdKafkaException.FromErr(LibRdKafka.last_error(), "Could not produce message");
+                }
+                return;


Use proper error string (rd_kafka_err2str(..last_error))

I'll put a todo for this, have a separate JIRA for sorting out exceptions / errors.

The risk of doing it outside the review process is that we might loose track of requested changes and that means we'll eventually end up having to re-review the entire code base.
If things are fixed in a followup commit in the same PR it is much easier to track.

Well, this class is going to be completely removed, so this is going to need to be re-reviewed at some point anyway - I'm in the middle of a big refactor - I see a lot of value in getting things broadly in place first then focussing on the detail. I see capturing things as todo's which get coppied around pretty efficient (keeping in mind my first point in this comment). Also, it's the best way for me to get a holistic view of the whole project, which I believe reduces risk in making bad design decisions. This is particularly important for me as i'm new to clients so can foresee less than I otherwise might.

edenhill · 2016-11-22T19:45:47Z

src/Confluent.Kafka/Handle.cs

+
+        public void Flush()
+        {
+            while (OutQueueLength > 0)


there is actually a rd_kafka_flush() call that should beused.

Yeah, I checked and this wasn't exposed in the LibRdKafka layer yet whereas the OutQueueLength stuff was. Agreed that we should use the correct internal version though.

Yep. I was just doing whatever ah- was here. will change.

I'll put a todo. I want to prioritize getting the high level API right, and i'm going to be reviewing / addressing a lot more lower level stuff in future PRs

edenhill · 2016-11-22T19:47:41Z

src/Confluent.Kafka/Producer.cs

-        {
-            producer.Dispose();
-        }
+        public bool FlushOnDispose => producer.FlushOnDispose;


I think we should really try to keep everything as config dict properties.

This seems very application code specific to me (users of an application shouldn't ever want to set it) and setting it in the config unnatural, so I think best left as a property. @ewencp ?

Yes, but this is what we do in librdkafka, Python and Go.
Since it is up to the application to allow users to set configuration properties it can also block these ones if it so desires (we could add a helper that assists in this: rd_kafka_conf_property_is_probably_not_for_the_user(str) bool).

But it's more difficult to use and not at all idimoatic the way you suggest.
And it should never be exposed outside the app (why should a user even know what Dispose is - it's a c-sharp thing - let alone how and when the app uses it).

edenhill · 2016-11-22T19:49:18Z

src/Confluent.Kafka/Serialization/IntDeserializer.cs

    public class IntDeserializer : IDeserializer<int>
    {
+        /// <remark>
+        ///     Endianness depends on architecture


The serializer must be stable and not dependent on arch.
If we have a producer on little endian and a consumer on big endian they need to be compatible using the same Serializer.

right you are.

mhowlett · 2016-11-22T23:00:23Z

examples/Wrapped/Program.cs

                // (null, int) serialization. When you do not wish to write any data to a key
                // or value, the Null type should be used.
-                var sProducer2 = producer.Wrap<Null, int>(null, new IntSerializer());
+                var sProducer2 = producer.Wrap<Null, int>(null, new IntSerializer(Endianness.LittleEndian));


I've made a new enum Edianness. Alternatively could use a bool here. This is more self-descriptive, but if the value is determined at runtime, could be a bit more annoying to use. opinions?

If the IntSerializer is only aimed at being compatible with itself then it must not have a configuration option to specify endian-ness, but instead be hardcoded to little or big endian.
Otoh if we think this Serializer will need to be compatible with IntSerializers in other languages we shall investigate if there is any prior art and if so adhere to that endian ness.
E.g., Avro uses little endian, Kafka uses big endian.

I dont really see the point of having the endian configurable, that'll create more problem than it solves.

good point. let's just make it network order.

edenhill · 2016-11-28T17:22:02Z

src/Confluent.Kafka/Handle.cs

+        // TODO: Add timout parameter (with default option == block indefinitely) when use rd_kafka_flush.
        public void Flush()
        {
+            // TODO: use rd_kafka_flush here instead..


It would've been less code calling rd_kafka_flush() than adding these comments ;)

edenhill · 2016-11-28T17:22:57Z

src/Confluent.Kafka/Serialization/IntDeserializer.cs

        public int Deserialize(byte[] data)
        {
-            return BitConverter.ToInt32(data, 0);
+            // network byte order -> big endian -> most significant byte in the smallest address.


There must be a system lib function for this (which also avoids doing anything on big-endian systems)

http://stackoverflow.com/questions/2420227/ntohs-and-ntohl-equivalent

What i'm wondering is if calling that then BitConverter will be faster or slower than what I've got, also considering I expect most people won't be running on big endian systems (i have no idea).

Something irks me about using this function that takes an int and returns and int, because what it's returning semantically isn't actually an int ... I'm also not convinced it's going to be faster in the the arithmetic expression I've got (i think it's fine to optimize for little endian systems).

Using System.Net.IPAddress.HostToNetworkOrder() also brings in a dependency on System.Net which is not currently required. I don't have a clear idea of what this means. It should always be on the host system because it's part of the platform, but it might mean a larger memory footprint (dll loaded when it otherwise wouldn't have been).

edenhill · 2016-11-28T17:28:25Z

src/Confluent.Kafka/Serialization/IntDeserializer.cs

-            return BitConverter.ToInt32(data, 0);
+            // network byte order -> big endian -> most significant byte in the smallest address.
+            return
+                (((int)data[0]) << 24) +


Assuming this is for a little-endian host, I think this might be wrong, it should be other way around:

return (data[3] << 24) | (data[2] << 16) | (data[1] << 8) | (data[0]);

Endianness of the host doesn't matter with << and >> operators.

I believe I have it the right way around.
Good point about the | operator though, that'll be quicker.

Ah, yes, you are right, sorry.

edenhill · 2016-11-28T17:47:46Z

src/Confluent.Kafka/DeliveryReport.cs

+    Same deal with Go where the DR channel gets one of these:
+    http://docs.confluent.io/3.1.0/clients/confluent-kafka-go/index.html#Message Is this being
+    kept more minimal intentionally?
+    */


As discussed on Slack, the message metadata contains an ever growing number of fields, so passing a rich Message object to the delivery report , like the other clients, is most likely the best way forward.
And that Message object should be the same as returned by consumer.poll()

right. however if a very common use case is to just produce a key and value (I think it is), then it's worth having a Producer.ProducerAsync overload for this as well to make the interface easier to use.

It is hard to make assumptions on what information the dr callback needs based on the produce() arguments. We should provide whatever librdkafka provides in the dr.

Producer Refactor Mk II

c57e904

mhowlett commented Nov 21, 2016

View reviewed changes

ewencp reviewed Nov 21, 2016

View reviewed changes

edenhill reviewed Nov 22, 2016

View reviewed changes

review changes

74cdb43

edenhill reviewed Nov 22, 2016

View reviewed changes

mhowlett commented Nov 22, 2016

View reviewed changes

Matt Howlett added 2 commits November 23, 2016 09:36

review II

acd0ab2

review III

6c7553b

edenhill reviewed Nov 28, 2016

View reviewed changes

edenhill suggested changes Nov 28, 2016

View reviewed changes

edenhill reviewed Nov 28, 2016

View reviewed changes

review IV

207db7f

edenhill approved these changes Nov 28, 2016

View reviewed changes

mhowlett merged commit b1eaf65 into confluentinc:master Nov 28, 2016

mhowlett deleted the producer3 branch November 29, 2016 16:43

koushikchitta mentioned this pull request Mar 7, 2018

Invalid offsets on successful delivery of message with 0.11.3 #453

Closed

7 tasks

nitinpi mentioned this pull request Jan 31, 2019

Unable to receive messages : Group coordinator not available #756

Closed

7 tasks

adimoraret mentioned this pull request Jan 18, 2024

Issues with API version negotiation between Kafka producer client and Kafka broker #2175

Open

8 tasks

		}


		public class Producer<TKey, TValue> : ISerializingProducer<TKey, TValue>, IDisposable

Producer Refactor Mk II #8

Producer Refactor Mk II #8

Uh oh!

Conversation

mhowlett commented Nov 21, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhowlett Nov 22, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhowlett Nov 22, 2016 •

edited

Loading