Ensure write and order guarantees with Producer.committableSink

[seglo]

References #1242

Enforce enable.idempotence when using the Producer.committableSink. This will ensure that produced messages are recorded once and in the order they were sent to the broker. Some caveats with this solution:

• The broker must be version 0.11 or greater
• max.in.flight.connections must be <= 5 (enforced by setting enable.idempotence=true)
• acks must be set to all (enforced by setting enable.idempotence=true)

[seglo]

Initial benchmarks indicate enforcing enable.idempotence with Producer.committableSink has little to no effect on producer throughput. It could be that other Alpakka Kafka performance penalties outweigh any performance loss provided by this setting, or it could be that our benchmark infrastructure scale is just not sufficient to observe a notable difference. Either way, if this fixes the dropped/unordered messages it may be an acceptable solution when using Producer.committableSink.

[seglo]

@PrathikPuthran The intention is that using enable.idempotence could resolve the issue you raised in #1242. Can you try this PR, or just add enable.idempotence to your ProducerSettings yourself?

[seglo]

@PrathikPuthran Hello again. Let me know if you have time to check if this setting resolves the issue for you. If it does then we can make this the default for 2.1.0.

[PrathikPuthran]

@seglo Sorry for the delay. Would this work if custom partitioning is used in the kafka producer?

[seglo]

@seglo Sorry for the delay. Would this work if custom partitioning is used in the kafka producer?

@PrathikPuthran Yes, I don't see why not. All it's doing is using sequence numbers (per destination partition for the length of the producer session) to ensure message ordering and for the broker to determine if any messages have been missed or duplicated. Missed messages will result in a fatal exception, dupes will be de-duped silently on the broker side. Your broker must be at least version 0.11.

[PrathikPuthran]

Yes but if send to one broker partition fails (if one broker is unhealthy) but the send for some other broker partition succeeds with messages of higher offsets you would still be committing offsets out of order I think?

[seglo]

Yes but if send to one broker partition fails (if one broker is unhealthy) but the send for some other broker partition succeeds with messages of higher offsets you would still be committing offsets out of order I think?

I did some digging and I can't find an official statement that callbacks can return out of order with enable.idempotence=true, but I think it's implied that this should not happen. If you can reliably reproduce this commit out of order issue I would encourage you to try it anyway. Otherwise it may take awhile for me to verify it myself.

[seglo]

This may be better served as a note in the docs on Producer.committableSink, instead of enforcing a Producer property.

[seglo]

If someone can confirm if enable.idempotence resolves the issue I'll refresh this PR with a docs update. Closing for now.