Kafkajs: Transactional producer

[tulios]

The transactional producer allows an application to send messages to multiple partitions (and topics) atomically. Each transactional producer has a unique transactionalId, which is used to enable transaction recovery across multiple sessions.

There can be only one open transaction per instance of the producer.

__Proposed API:__

const producer = kafka.producer({ transactional: true })

const run = async () => {
  await producer.connect()
  const tx = await producer.tx()

  try {
    for (let i = 0; i < 100; i++) {
      await tx.add({
        topic,
        messages: [{ key: `key-${i}`, value: `value-${i}` }]
      })
    }

    await tx.commit()
  } catch (e) {
    // Investigate class of errors where the producer can't recover, i.e: ProducerFencedException
    await tx.abort()
  }

  await producer.disconnect()
}

__Workflow:__

await producer.connect():

• Should call initProducerId to initialize the producer, since it can only happen once per instance it could be combined with connect
• Add a new log line to indicate the use of a transactional producer

const tx = await producer.tx():

I still have mixed feeling about this design, but the idea is to create a transaction object where the user will be able to add records, commit, and abort.

• It should check for other producers with the same ID
• In memory flag to indicate that a transaction was created. There can be only one open transaction per instance of the producer.

await tx.add({ ... }):

Adds the records to some record accumulator

await tx.commit():

• Calls AddPartitionsToTxn
• Calls EndTxnRequest with COMMIT

await tx.abort():

• Clear the in-memory accumulator
• Calls EndTxnRequest with ABORT

We have to investigate the errors that cause the consumer to disconnect instead of abort. From the Java client, we have: ProducerFencedException, OutOfOrderSequenceException, and AuthorizationException where the authorization error is specific to the current transactionalId

await producer.disconnect():

Should close the connection as usual

__Open questions:__
Do we need guidelines for the transactionalId?

Should we auto-generate the transactionId if it's not defined?

__NOTE__:

The API design is still a draft

Depends on #165 #166 #167 #170

@Nevon @ianwsperber ^

[Nevon]

Some thoughts from our discussion.

tx.add could be tx.send if we send the messages to the server, rather than buffering them up on the client. This would align the transaction API with the producer API nicely. abort would then become rollback.

An alternative way to create transactions could be something like:

await producer.tx(async (tx) => {
  await tx.send(...)

  // to rollback, throw
})

Not sure if this would lead to more complicated code for the user.

Another thing I'd like to think about is if we need this transactional flag when creating the transactional producer. We could either create a separate factory method, like transactionalProducer, or make the producer transactional when creating a transaction (i.e. the first call to producer.tx).

[ianwsperber]

I think instead of a transactional flag we could provide a transactionId in the producer instantiation. Since it's used for fencing maybe we shouldn't try to autogenerate it if it's missing?

I'm curious to hear the justification for returning a transaction over making the transaction methods available on the producer. It's not strictly necessary (the Java API makes all the methods available on the producer) so would just like to better understand why it'd be an improvement.

@Nevon IIRC Knex uses the same callback design for transactions. Personally I've always found it hard to work with, an awkward mix of promises and callbacks.

[tulios]

The idea behind the "transaction" object is to improve the developer experience, the approach used by the Java API is error-prone, you might call the methods in the wrong order and get weird errors. With the transaction object, we could remove the need of some of the methods by initializing things on connection, etc. On the other hand, it exposes a different API so if users are used to the Java API it might cause confusion but I still believe it should be idiomatic, we shouldn't blindly follow the Java API.

But I'm open to any design. Since @ianwsperber is moving fast with the EoS we should try to finalize the API and also spec the idempotent producer (which is part of the transactional producer anyway)

[ianwsperber]

@tulios I think you have a fair point in not wanting to blindly follow the Java API, it's certainly possible to write something clearer & more concise.

Since the idempotent producer is requisite for the transactional producer do you think we ought to have a separate ticket for its implementation? AFAIK the only additional protocol the idempotent producer requires is InitProducerId, which we've already implemented (#166), and the record batch which exists in the repo already. It could be released prior to the rest of the transactional work.

[tulios]

I think you can finalize the protocol parts; you are almost there.

[ianwsperber]

@tulios With the protocols now implemented I'm thinking we can move forward with an idempotent producer implementation. IMO we should allow users to create an idempotent producer regardless of whether they'll participate in a transaction. We could accomplish this with an additional idempotent: boolean flag in the producer options. Thoughts?

[tulios]

I've checked other libraries, and everybody is following this pattern. Let's follow suit.

[ianwsperber]

@tulios Created #200 so we can separately track the idempotent producer work

[ianwsperber]

@tulios With #203 in PR I'm going to move forward implementing this following the API outlined above. If we want to make any changes I think that can be handled in PR, the implementation should be mostly independent of the external API.