Spring Kafka Wait For Assignment Failed After Update To SpringBoot 3.2.0

[dimasadryantos]

Problem :

• I have test where I configure multiple topics and have test container waiting for assignments, after updating to SpringBoot 3.2.0, my integration test freezing and returning :

Caused by: java.lang.IllegalStateException: Expected 2 but got 1 partitions

    @EventListener(ContextRefreshedEvent::class)
    fun waitForAssignment() {
        for (messageListenerContainer in kafkaListenerEndpointRegistry.listenerContainers) {
            if (shouldWaitForAssignment(messageListenerContainer)) {
                ContainerTestUtils.waitForAssignment(messageListenerContainer, embeddedKafka.partitionsPerTopic)
            }
        }
    }

[scottfrederick]

Thanks for getting in touch. Please try downgrading your Spring Kafka dependency to 3.0.13 to match the version that was managed with Spring Boot 3.1.x. You can do this by setting the spring-kafka.version property in your build. This will help us to know if the problem is in Spring Boot or Spring Kafka.

If that doesn't change the problem, please provide a complete minimal sample that reproduces the problem. You can share it with us by pushing it to a separate repository on GitHub or by zipping it and attaching it to this issue.

[PospolitaNV]

I have the same issue

[dimasadryantos]

@scottfrederick Hi Thank you for your feedback, yes I will attach sample project for reproducing the issue

[ThomHurks]

In my integration tests I had the same issue after upgrading to 3.2, I worked around it by making sure an integration test is for a single topic and has partitions set to 1 on the embedded Kafka annotation. It seems like a weird race condition, because when I attached a debugger I could see the number of expected partitions was correct, but when just running the test it would fail with the same error as reported here.

[dimasadryantos]

Hi @ThomHurks could you add example project to reproduce the error? As I tried with my project the error didn't appear

[spring-projects-issues]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[spring-projects-issues]

Closing due to lack of requested feedback. If you would like us to look at this issue, please provide the requested information and we will re-open the issue.

[rgolder1]

More information on my observation of this change in behaviour:
https://github.com/lydtechconsulting/introduction-to-kafka-with-spring-boot/wiki#error-running-integration-tests-with-spring-boot-32
I am unclear if the change is a bug, or a fix to what was a bug, or just a functional change in behaviour.

Steps to reproduce:

• clone the above project
• switch to branch 13-dlt
• run the Spring Boot test OrderDispatchIntegrationTest - test passes
• change the Spring Boot version in the pom.xml from 3.0.5 to 3.2.1
• run the Spring Boot test OrderDispatchIntegrationTest - test fails with error:
  java.lang.IllegalStateException: Expected 2 but got 1 partitions

[wilkinsona]

Thanks, @rgolder1. The problem does not occur with Spring Boot 3.2.1 when Spring Kafka is downgraded to 3.0.13 by adding <spring-kafka.version>3.0.13</spring-kafka.version> to the pom's <properties> so the change in behavior appears to be do to a change in Spring Kafka.

/cc @artembilan who maintains Spring Kafka

[artembilan]

Thank you for great example to play with!

Although it is too complicated to digest quickly, but here is a workaround for now:

@EmbeddedKafka(controlledShutdown = true, kraft = false)

Starting with Spring for Apache Kafka 3.1, the embedded Kafka is switched to a new KRaft model: https://developer.confluent.io/learn/kraft/.

Apparently it does not take into account a number of partitions set by default in the @EmbeddedKafka.partitions attribute when those topics are created automatically.
Another workaround is like:

@EmbeddedKafka(controlledShutdown = true, partitions = 1)

or more fine-grained:

@EmbeddedKafka(controlledShutdown = true, topics = { OrderDispatchIntegrationTest.ORDER_CREATED_TOPIC, OrderDispatchIntegrationTest.DISPATCH_TRACKING_TOPIC, OrderDispatchIntegrationTest.ORDER_DISPATCHED_TOPIC, OrderDispatchIntegrationTest.ORDER_CREATED_DLT_TOPIC })

to make an embedded Kafka to create those topics for us upfront.

As for the problem with KRaft.
I see there is a logic in the EmbeddedKafkaZKBroker:

				if (!this.brokerProperties.containsKey(KafkaConfig.NumPartitionsProp())) {
					brokerConfigProperties.setProperty(KafkaConfig.NumPartitionsProp(), "" + this.partitionsPerTopic);
				}

So, we propagate a specific property into Kafka broker started via embedded Kafka for topics auto-creation.
Doesn't look like we have similar logic in the EmbeddedKafkaKraftBroker:

	private void addDefaultBrokerPropsIfAbsent(Properties brokerConfig, int numBrokers) {
		brokerConfig.putIfAbsent(KafkaConfig.DeleteTopicEnableProp(), "true");
		brokerConfig.putIfAbsent(KafkaConfig.GroupInitialRebalanceDelayMsProp(), "0");
		brokerConfig.putIfAbsent(KafkaConfig.OffsetsTopicReplicationFactorProp(), String.valueOf(numBrokers));
	}

So, feel free to raise a respective issue in the https://github.com/spring-projects/spring-kafka/issues.

Or, @wilkinsona , we can just transfer this issue over there.