Rebalancing problem between Spring Kafka consumers with versions <=2.3.13.RELEASE and >=2.4.0.RELEASE (included 2.8.6)

[ferblaca]

I am having problems with kafka consumers of applications with different versions of Spring-kafka, specifically between 2.3.13.RELEASE and >=2.4.0.RELEASE (included 2.8.6).

We have noticed that when performing a blue/green deployment, in which we upgrade consumer applications that were on Spring-Kafka version 2.3.13.RELEASE to [2.4.0.RELEASE,2.8.6], and that for a while both versions of the applications are deployed and active at the same time, the applications with Spring-Kafka version [2.4.0.RELEASE,2.8.6] do not get any partitioning under the following circumstances:

• Both applications have a consumer with the same group name to the same topic.
• Both applications have the consumer configured with concurrency equal to the number of partitions the topic has.

The result is that, for example, if the topic has 10 partitions, the application with Spring-Kafka version 2.3.13.RELEASE is assigned all 10 partitions, while the application with Spring-Kafka [2.4.0.RELEASE,2.8.6] is not assigned any partition.

No matter in which order the applications are deployed, the end result is always the same.

When the Spring-Kafka 2.3.13.RELEASE application stops, the [2.4.0.RELEASE,2.8.6] application is assigned all 10 partitions correctly.

For more detailed information on application logs, see the following link.

Would it be possible by configuration to avoid this behaviour?

I attach the apps that reproduce this described behavior: springKafkaIssue.zip

[garyrussell]

This has nothing to do with Spring; all rebalancing activity is handled by Kafka itself.

Recent versions of the kafka-clients changed the partition assignor; in your first log

partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]

In the second log:

 partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor, class org.apache.kafka.clients.consumer.CooperativeStickyAssignor]

The sticky assignor tries to keep partitions with the same instance.

[ferblaca]

@garyrussell I have tried forcing the RangeAssignor strategy for both versions:

partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor].

with the same result.

On the other hand, I have also tested without Spring, only with the Kafka-client for the implied versions simulating concurrency and it didn't reproduce either.

[garyrussell]

Although, according to their docs for changes in the 3.0.0 client:

The default partition.assignment.strategy is changed to "[RangeAssignor, CooperativeStickyAssignor]", which will use the RangeAssignor by default, but allows upgrading to the CooperativeStickyAssignor with just a single rolling bounce that removes the RangeAssignor from the list. Please check the client upgrade path guide here for more detail.

https://kafka.apache.org/documentation/#consumerconfigs_partition.assignment.strategy

So it's not clear why the sticky assignor is being used.

[garyrussell]

I'll play with your example to see if I can see what's going on, but I don't see how it can be a Spring issue.

[garyrussell]

But, in general, as Tomaz replied on SO, Kafka just sees 20 consumers; he doesn't know that they are 2 sets of 10, so there is no guarantee that each instance will get 5; it's generally better to over-provision the number of topics for this kind of situation.

[ferblaca]

But, in general, as Tomaz replied on SO, Kafka just sees 20 consumers; he doesn't know that they are 2 sets of 10, so there is no guarantee that each instance will get 5; it's generally better to over-provision the number of topics for this kind of situation.

yes.. but in all the tests I have performed, which have been many, the partitions are always evenly distributed (5 partitions for each consumer of each App), as long as the Spring-kafka versions of the applications are not those mentioned above.

The tests performed only with the kafka-client (without Spring), creating 10 concurrent consumers for each App for 2.4.1 and 3.1.0 versions of kafka-client does not reproduce in any case the behavior described in the issue, that is to say, always in all cases an equal distribution of partitions for each App is performed.

It is a strange behavior that I can't understand which I have been able to reproduce from a local environment until we are managed by K8s.

[garyrussell]

It is strange, if I run 2 copies of consumer1 or 2 copies of consumer 2 each gets 5 partitions. But if I run one of each, then consumer1 retains all 10.

However, I see the same behavior when there is no Spring involvement; I changed your mains as follows:

public static void main(String[] args) throws InterruptedException {
	// SpringApplication.run(DemoApplication.class, args);
	Map<String, Object> config = Map.of(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092",
			ConsumerConfig.GROUP_ID_CONFIG, "foo",
			ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
			ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
	SimpleAsyncTaskExecutor exec = new SimpleAsyncTaskExecutor();
	for (int i = 0; i < 10; i++) {
		exec.execute(() -> {
			KafkaConsumer consumer = new KafkaConsumer(config);
			consumer.subscribe(List.of("spring.kafka.test"), new ConsumerRebalanceListener() {

				@Override
				public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
					log.info("Revoked: " + partitions);
				}

				@Override
				public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
					log.info("Assigned: " + partitions);
				}
			});
			while (true) {
				consumer.poll(Duration.ofSeconds(5));
			}
		});
	}
	Thread.sleep(60_000);
	System.exit(0);
}

With the same result; consumer1:

11:17:27.212 [SimpleAsyncTaskExecutor-6] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-3]
11:17:27.212 [SimpleAsyncTaskExecutor-4] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-8]
11:17:27.212 [SimpleAsyncTaskExecutor-5] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-5]
11:17:27.212 [SimpleAsyncTaskExecutor-9] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-2]
11:17:27.212 [SimpleAsyncTaskExecutor-10] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-1]
11:17:27.212 [SimpleAsyncTaskExecutor-7] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-9]
11:17:27.212 [SimpleAsyncTaskExecutor-8] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-4]
11:17:27.212 [SimpleAsyncTaskExecutor-3] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-0]
11:17:27.212 [SimpleAsyncTaskExecutor-2] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-6]
11:17:27.212 [SimpleAsyncTaskExecutor-1] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-7]
11:17:33.211 [SimpleAsyncTaskExecutor-2] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-6]
11:17:33.212 [SimpleAsyncTaskExecutor-4] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-8]
11:17:33.212 [SimpleAsyncTaskExecutor-7] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-9]
11:17:33.212 [SimpleAsyncTaskExecutor-6] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-3]
11:17:33.213 [SimpleAsyncTaskExecutor-9] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-2]
11:17:33.213 [SimpleAsyncTaskExecutor-1] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-7]
11:17:33.213 [SimpleAsyncTaskExecutor-8] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-4]
11:17:33.214 [SimpleAsyncTaskExecutor-5] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-5]
11:17:33.215 [SimpleAsyncTaskExecutor-3] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-0]
11:17:33.216 [SimpleAsyncTaskExecutor-10] INFO  c.s.kafka.sample.DemoApplication - Revoked: [spring.kafka.test-1]
11:17:33.239 [SimpleAsyncTaskExecutor-5] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-5]
11:17:33.239 [SimpleAsyncTaskExecutor-10] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-1]
11:17:33.239 [SimpleAsyncTaskExecutor-8] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-4]
11:17:33.239 [SimpleAsyncTaskExecutor-9] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-2]
11:17:33.239 [SimpleAsyncTaskExecutor-6] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-3]
11:17:33.239 [SimpleAsyncTaskExecutor-7] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-9]
11:17:33.239 [SimpleAsyncTaskExecutor-1] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-7]
11:17:33.239 [SimpleAsyncTaskExecutor-3] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-0]
11:17:33.239 [SimpleAsyncTaskExecutor-4] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-8]
11:17:33.240 [SimpleAsyncTaskExecutor-2] INFO  c.s.kafka.sample.DemoApplication - Assigned: [spring.kafka.test-6]

Consumer2:

11:17:33.242 [SimpleAsyncTaskExecutor-2] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []
11:17:33.242 [SimpleAsyncTaskExecutor-3] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []
11:17:33.242 [SimpleAsyncTaskExecutor-8] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []
11:17:33.242 [SimpleAsyncTaskExecutor-9] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []
11:17:33.242 [SimpleAsyncTaskExecutor-5] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []
11:17:33.242 [SimpleAsyncTaskExecutor-10] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []
11:17:33.242 [SimpleAsyncTaskExecutor-1] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []
11:17:33.242 [SimpleAsyncTaskExecutor-7] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []
11:17:33.242 [SimpleAsyncTaskExecutor-6] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []
11:17:33.242 [SimpleAsyncTaskExecutor-4] INFO  c.s.kafka.sample.DemoApplication2 - Assigned: []

So I still don't consider this to be a Spring issue.

[garyrussell]

I corrected my test to put the poll in a loop; but I still get the same results - edited the comment above.