Add partitioned transactional source phase 1

[seglo]

Purpose

This PR adds a new partitioned source with support for Kafka transactions, Transactions.partitionedSource. A Source is created per partition in a consumer group subscription. Each Source must have a downstream Transactional.sink|flow to commit messages transactionally. A derived transactional.id is used per Transactional.sink|flow that's based on a user-provided transactional.id along with the group.id, topic and partition of the originally consumed message. Transactions.partitionedSource will make it easier to run transactional workloads across consumer groups with multiple members.

This PR is based on the extensive work already completed by @charlibot earlier this year. It was rebased by @raboof and updated to accommodate transactional consistency work completed by @2m and @szymonm. This new branch will serve as the integration point to get this functionality into Alpakka Kafka 2.0.0.

Update 27/11/2019

Due to the large number of changes introduced by this PR it has been repurposed as a "phase 1" to lay the groundwork to continue this work. This PR will not expose the Transactional.partitionedSource to the end user.

References

• Original PR: Add partitioned transactional source #705
• Transactions draining logic PR: Implement draining logic for transactional flow #757
• Original Transactions PR: Kafka Transactions #420

Phase 1 tasks

• Core impl.
• Confirm producer per substream is acceptable (Add partitioned transactional source #705 (comment))
• Review docs. Include details about KIP-447.
• Multi-broker integration test (blocked by Use testcontainers for multi-broker integration tests and benchmarks #939)
• Test: If sub source fails, rebalance occurs and new source is created (Add partitioned transactional source #705 (comment))
• Test: Consistency after rebalance of regular transactions
• Update docs. (to be continued in phase 2)
• Test retry infrastructure

Phase 2 tasks (preliminary)

• Test: Consistency of partitioned source when calling shutdown using draining control
• Set ConsumerSettings.stopTimeout to 0 to allow for faster shutdowns when a stream completes/fails.
• Update docs. (to be continued in phase 2)
• Add more low level testing of draining control logic

[seglo]

I received confirmation that Kafka Streams EoS creates one producer per source partition. This will introduce performance implications which will be addressed in KIP-447, Producer scalability for exactly once semantics. This work is in progress and will land in a future version of Kafka (not 2.4.0).

• Talk: Revisiting Exactly One Semantics (EOS) by Jason Gustafson
• kafka-users thread, Questions about Producer per Task/Partition in Streams EoS Impl.

In the meantime we can proceed with our current implementation to have one producer per partitioned source.

[ennru]

Many places would improve with named classes.

[ennru]

Quite a chunk of code.
I'm not sure what is tested and what is not. We should find a way to document that, especially when we move things across integration tests and "fast" tests.

Sprinkle more final on classes.

[ennru]

I commented it out and added final, private, and protected at a few places.

[ennru]

LGTM.