Add a way to bypass batching in `Writer.WriteMessages`

[nlsun]

It appears that there's demand for bypassing the batching in Writer.WriteMessages judging from the comments in https://github.com/segmentio/kafka-go/issues/852#issuecomment-1243061407

I can think of roughly 3 different ways to go about this:

1. We can add a set of Writer flags that disables functionality (e.g. batching) as needed
2. We can add it to the Writer as a dedicated function like in this PR.
3. Are the Client and Balancer all that's needed? Maybe we could write a snippet in Examples that shows a user how to craft their own function?

[nlsun]

@yusufozturk thinking we can continue the discussion here since it has diverged a little bit from the original topic in that other issue.

The PR you submitted looks similar to option 2 if I understand it correctly.

Given the options I noted above (and feel free to suggest more) I think I might prefer 1 or 3 over option 2. Just because Option 2 will expand what functions we need to maintain. Option 1 does add some complexity but looking at the code it doesn't seem horrendous. And Option 3 is by far the simplest if this is considered an "uncommon and advanced" use case. I actually slightly prefer Option 3 unless we find out that this is a very common use case.

[achille-roussel]

I agree that the change in https://github.com/segmentio/kafka-go/pull/986 appears to demonstrate that maybe the implementation is simple enough that it's not worth providing as part of the kafka.Writer.

Partition selection based on the first message passed to the message seems strongly opinionated and could result in unexpected behavior (e.g. what if the messages were intended to go to different partitions). It seems in those cases the application has to have the responsibility of deciding which partition to route the batch to anyways, so probably better to make it explicit in the code rather than reusing abstractions which were not designed to support this use case.

I'd support starting with a code example, then we can figure out what APi needs to be added to facilitate partitioning of batches in pair with a kafka.Client.

[3AceShowHand]

How about set the BatchSize to 1, this should skip the batch.

[beebeeep]

How about set the BatchSize to 1, this should skip the batch.

this results in sending messages one by one, even if you provided big batch to Writer.WriteMessage(). I.e. if you write 100 messages, Writer will do 100 ProduceRequest's of 1 message instead of 1 request with batch of 100.