Transactional Producer (full EOS, KIP-98)

[edenhill]

This is work in progress.

Current status:

• ✔️ Functionally complete - try it out in action!
• ✔️ Functional tests are complete.
• ✔️ Public API is settled.
• ✔️ Proper error handling across, including controller switches during transactions
• 🚧 API docs and manual are not done and will be finalized in a sub-sequent PR.

Review guide:

• The transactional producer relies on the idempotent producer: to acquire the initial ProducerID, and for the message guarantees. Most fatal errors in the idempotent producer are only fatal to the current transaction when using the transactional producer, resulting in the ABORTABLE_ERROR state.
• Most in-transaction errors, be they from the underlying idempotent producer, or failure to add offsets or partitions to the transaction, result in an Abortable Error, which is a transactional error state that requires the application to call abort_transaction() and then start over again with begin_transaction().
• All the transactional logic is the transaction manager, rdkafka_txnmgr.c, with some leaky abstraction in rdkafka_idempotence.c for the interplay between the idempotent producer and the transaction manager.
• The use of coord_req() (rdkafka_coord.c), which is an abstraction on top of coordinator brokers makes it easier to perform retries, etc. External community work on the Admin API has already started using it (based on the txns branch).
• Matt's 0098 transactional consumer test has been extended to run identical tests with the Java txn producer and the librdkafka txn producer.
• Ignore documentation FIXMEs for now, docs will be finalized after this PR has been merged.

Miscellaneous:

• Adds a builtin mock cluster with error injection.
• Manual unittest code coverage functionality as an alternative to object inspection.

[mhowlett]

'setting in the broker' -> 'setting on the broker' i think

[mhowlett]

I think you probably want to do a search for 'FIXME'.

[mhowlett]

this is really dangerous - people are going to screw this up by default because everyone expects committed offset to be the last messages successfully read (and it's what you naturally have in your code without doing any arithmetic, which people won't do by default unless they're thinking hard about it, which they won't be). when offsets are committed using the consumer, this isn't such a big deal because the consumer does the calculation for you behind the scenes (for most variants). But in this case, the user will be in the loop.

given this, i think in the higher level bindings we should have an overload of SendOffsetsToTransaction which accepts a consumer instance. the java producer doesn't have this.

[edenhill]

I hear you, but we have existing offset interfaces that take the actual offset (commit, store_offsets), without doing +1, so we need to be careful with being consistent in this regard.

Best thing would be if users did not have to care about individual partitions and offsets, but instead, as you propose, provide a helper method taking a consumer.

In fact, we really need three things:

1. set start-of-transaction snapshot/restore point
2. commit/send-offsets for transaction
3. rewind to snapshot/restore point on transaction abort

Which would allow you to do something along the lines of:

while the_truth:
    producer.begin_transaction()
    consumer.set_restorepoint()

  try:
    while time_in_txn < max_time_per_txn:
       input = consumer.poll(remaining_time)

       if consumer.is_rebalancing():
           producer.abort_transaction()
           consumer.clear_restorepoint()
            continue outer_loop

       output = process(input)
       producer.produce(output)

    producer.commit_offsets(consumer)
    producer.commit_transaction()

  except AbortableError:
          producer.abort_transaction()
          consumer.rewind_to_restorepoint()
          continue outer_loop

[rnpridgeon]

It would be nice to expose rd_kafka_is_transactional for such cases so we can fail fast on eos consumer instantiation.

[rnpridgeon]

It's not terribly clear to me from the mega design doc what happens in the event of a rebalance when a transaction is still open. Do we need to abort as implied by the snipped above, or is there a way for a consumer to pick up the last offset sent to a transaction for a given partition upon assignment.

[rnpridgeon]

Mystery solved

https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics

[edenhill]

It would be nice to expose rd_kafka_is_transactional for such cases so we can fail fast on eos consumer instantiation.

Can you elaborate on this? Where and how would it be used?

[mhowlett]

haven't looked at the code yet to check actual behavior, but I think this should fail immediately if there are any logic offsets (and this docstring updated accordingly).

[edenhill]

The idea here is that you should be able to pass the current position() directly to this function, and some partitions may not have been consumed at all (due to emptyness, no messages, paused, whatever) and thus may have a logical current position.

[alex88]

🎉 glad to see this merged! Thanks for the awesome work!