Idempotent Producer

[ianwsperber]

Allow users to create idempotent producers. We create an idempotent producer by passing an idempotent flag during creation:

const idempotentProducer = kafka.producer({ idempotent: true });

Creating an idempotent producer is a prerequisite for the transactional producer (see related discussion in #173 )

Tasks

• Generate PID using protocol InitProducerId v0 during initialization (perhaps when we call producer.connect())
• Track sequence numbers per TopicPartition
• Send the PID, epoch and sequence with each record batch (see work already done in Add transactional attributes to record batch #199)
• Ensure we enforce acks, retries & max requests (or just log a warning...). Partially blocked by Kafka Consumer - handler concurrency limit? #148

Spec

From KIP-98:

When idempotence is enabled, we enforce that acks=all, retries > 1, and max.inflight.requests.per.connection=1. Without these values for these configurations, we cannot guarantee idempotence. If these settings are not explicitly overidden by the application, the producer will set acks=all, retries=Integer.MAX_VALUE, and max.inflight.requests.per.connection=1 when idempotence is enabled.

[ianwsperber]

@tulios Is there a configuration like the Java API's max.in.flight.requests.per.connection in place currently for KafkaJS? As far as I can tell there is not, though I haven't spent a lot of time looking through the networking or send messages code so don't want to make any assumptions. From what I can tell this would require implementing some sort of internal request queue.

[tulios]

KafkaJS doesn't have maxInflight connection yet, I might start it in 1 or 2 weeks. Take a look at #148 where we had a discussion about it.

[ianwsperber]

@tulios Ok cool. I don't think that's a blocker for implementing the rest of the idempotent producer, though it seems we can't strictly guarantee eos without it

[ianwsperber]

FYI I'm creating a "transactionManager" module for the producer, so we can easily share state around transactions. You can see how I'm planning to use it here, where I provide it to the sendMessages factory method so that it can later be used to increment the topic/partition sequence: https://github.com/tulios/kafkajs/blob/eos-200-idempotent-producer/src/producer/index.js#L27