Kafka cluster example

[seglo]

This PR introduces the KafkaContainerCluster to run a cluster of KafkaContainers. I discussed this with @bsideup on slack and thought it would be useful for others too.

Cluster-related changes

• Added convenience class KafkaContainerCluster to facilitate creating a multi-broker cluster.
• Implement Startable and use deepStart to launch cluster containers.
• Set appropriate replication factor for internal Kafka topics based on number of available brokers.
• Set broker id and broker id network aliases.
• Use Kafka AdminClient to create topics in tests so we can define topics with partitions and replication factors greater than 1.

Other changes

• Add default Kafka transaction topic replication factor to 1 so that users can run transactional workloads with a single broker (fixes KafkaContainer transactions: Timeout expired while initializing transactional state in 60000ms. #1816).
• Replace usage of deprecated KafkaConsumer.poll(int) with KafkaConsumer.poll(Duration) in tests.

[seglo]

Awesome, thanks a lot for the review @bsideup. I reverted most of the changes to KafkaContainer.

Is KafkaContainerCluster something you're willing to consider adding to the API? It should make it easier for users to setup multi-broker clusters without knowing much about Kafka.

[seglo]

Rebased on master. @bsideup LMK if you have any concerns.

[bsideup]

Have you considered using zookeeper.execInContainer(cmd)?

Also, what would be the use of this?

[seglo]

It's used in start() to assert that Kafka cluster has formed, by checking ZooKeeper (its coordinator) to see if the right number of brokers have joined together.

Good idea about using the ZooKeeper container. It looks like the zookeeper-shell script is present on the path there as well. I've pushed an update.

[bsideup]

I believe it can be simplified to () -> clusterBrokersIDs().length == brokersNum (where clusterBrokersIDs returns an array (or list) of broker IDs (or, if execInContainer is possible, can be inlined since it becomes a one liner)

[seglo]

The container command returns a string (internally it's stored as a string, I guess, i.e. 1,2,3). I could inline a shell command to split or count the string instead if that's what you prefer.

[bsideup]

I just realized that there is no error handling here, which means that if the start fails then the containers will be dangling until someone calls #stop().

That made me think of the rules use cases (some people still got used to running things with JUnit's rules) whether we should have it as part of the module at all, or perhaps we should make it an example instead (We have the examples/ folder for things that feel more like a very special use case rather than a thing that most of the users will be doing)

@seglo I remember that you were using it in Alpakka.
Given some cleanups we were able to do, do you still feel that this amount of code is preferable to have as a part of the Kafka module (and being maintained), or maybe you're okay with having it as an example instead?

P.S. the KafkaContainer's rf fix is still very much needed 👍

[seglo]

@bsideup By error handling do you mean if the containers fail to start?

I don't mind scrapping this PR (creating a separate one for the tx RF stuff), and making this code an example. As you know I have a copy in Alpakka that works well for us, though I would probably port it to Scala instead. LMK.

[bsideup]

@seglo

By error handling do you mean if the containers fail to start?

When some containers fail to start, or when the Unreliables part times out

I don't mind scrapping this PR (creating a separate one for the tx RF stuff), and making this code an example.

No need to, you can just adjust this PR! (Well, unless you want to make 2 separate contributions, totally deserved 👍)
P.S. in the example, I guess you can just define a few containers and add the Unreliables cluster check in @BeforeTest or something.

As you know I have a copy in Alpakka that works well for us, though I would probably port it to Scala instead

Let's do it like that: we will keep it as an example for now (I feel sorry for not suggesting it on a first place, btw! ☺️), and you will be able to evaluate the approach for a little longer.
I will monitor the activity and if we see that the use case is generic enough - we will move it into the module. How does it sound?

P.S. thanks a lot for the fast replies and your dedication! Really appreciate that 👍 💯

[seglo]

Thanks @bsideup . I'm having a gradle dep issue perhaps you can help me resolve. I added a build.gradle for ./examples/kafka-cluster, but when I attempt to run the example's tests I get dependency issues on testcontainer types, but I'm not sure why. I'm executing ../gradlew :kafka-cluster:test from within the examples directory:

 seglo@slice  ~/source/testcontainers-java/examples   seglo/multi-broker-kafka  ../gradlew :kafka-cluster:test

> Task :kafka-cluster:compileJava FAILED
/home/seglo/source/testcontainers-java/examples/kafka-cluster/src/main/java/com/example/kafkacluster/KafkaContainerCluster.java:27: error: cannot find symbol
    private final Network network;
                  ^
  symbol:   class Network
  location: class KafkaContainerCluster
/home/seglo/source/testcontainers-java/examples/kafka-cluster/src/main/java/com/example/kafkacluster/KafkaContainerCluster.java:28: error: cannot find symbol
    private final GenericContainer zookeeper;
                  ^
  symbol:   class GenericContainer
  location: class KafkaContainerCluster
/home/seglo/source/testcontainers-java/examples/kafka-cluster/src/main/java/com/example/kafkacluster/KafkaContainerCluster.java:29: error: cannot find symbol
    private final Collection<KafkaContainer> brokers;
                             ^
  symbol:   class KafkaContainer
  location: class KafkaContainerCluster
/home/seglo/source/testcontainers-java/examples/kafka-cluster/src/main/java/com/example/kafkacluster/KafkaContainerCluster.java:69: error: cannot find symbol
    public Network getNetwork() {
           ^
  symbol:   class Network
  location: class KafkaContainerCluster
/home/seglo/source/testcontainers-java/examples/kafka-cluster/src/main/java/com/example/kafkacluster/KafkaContainerCluster.java:73: error: cannot find symbol
    public GenericContainer getZooKeeper() {
           ^
  symbol:   class GenericContainer
  location: class KafkaContainerCluster
/home/seglo/source/testcontainers-java/examples/kafka-cluster/src/main/java/com/example/kafkacluster/KafkaContainerCluster.java:77: error: cannot find symbol
    public Collection<KafkaContainer> getBrokers() {
                      ^
  symbol:   class KafkaContainer
  location: class KafkaContainerCluster
/home/seglo/source/testcontainers-java/examples/kafka-cluster/src/main/java/com/example/kafkacluster/KafkaContainerCluster.java:87: error: cannot find symbol
    private Stream<GenericContainer> allContainers() {
                   ^
  symbol:   class GenericContainer
  location: class KafkaContainerCluster
/home/seglo/source/testcontainers-java/examples/kafka-cluster/src/main/java/com/example/kafkacluster/KafkaContainerCluster.java:105: error: cannot find symbol
    private String clusterBrokers(GenericContainer c) {
                                  ^
  symbol:   class GenericContainer
  location: class KafkaContainerCluster
8 errors

FAILURE: Build failed with an exception.

* What went wrong:
Execution failed for task ':kafka-cluster:compileJava'.
> Compilation failed; see the compiler error output for details.

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug option to get more log output. Run with --scan to get full insights.

* Get more help at https://help.gradle.org

BUILD FAILED in 2s
1 actionable task: 1 executed

[bsideup]

@seglo I see that you defined it as implementation, not testImplementation, probably that's the reason?

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. If you believe this is a mistake, please reply to this comment to keep it open. If there isn't one already, a PR to fix or at least reproduce the problem in a test case will always help us get back on track to tackle this.

[bsideup]

@seglo you won't believe! 😂 I just merged your PR into our own branch (for some polishing) and will merge it into master in a moment! Thanks A LOT for submitting it and staying with us :D And I am so sorry that it took so long 😅

[seglo]

Nice, thanks. Longest PR ever :)

I'll admit that I've been iterating this more over in Alpakka Kafka. I can take a look at the end result and see if there is anything worth carrying over.

[bsideup]

@seglo the longest indeed 😅 And with such a nice id, btw ;)

Looking forward to the updates, and I promise it won't take that much this time 😂