Share Consumer Example

[hantsy]

I tried to test the new Share Consumer feature, but after reading the doc, and created an example, I can not make it work as expected.

The example project: https://github.com/hantsy/spring7-sandbox/tree/master/boot-kafka-shared-consumer

My example is based on Spring Boot 4.1.0-RC1, and I set the kafka.version to 4.2.0 to use the latest Kafka clients, and also use kafka:latest docker image to ensure Kafka 4.2 in testcontainers.

The configuration class:

@Configuration
@Slf4j
class ShareConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    String bootstrapServers;

    @Bean
    NewTopic myTopic() {
        return new NewTopic(DEMO_TOPIC_NAME, 1, (short) 1);
    }

    @Bean
    public ShareConsumerFactory<String, String> shareConsumerFactory() {
        log.debug("Get bootstrap servers from properties:{}", bootstrapServers);
        Map<String, Object> props = Map.of(
                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,
                ConsumerConfig.GROUP_ID_CONFIG, DEMO_GROUP_NAME,
                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class
                //ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, "explicit"
        );
        DefaultShareConsumerFactory<String, String> factory = new DefaultShareConsumerFactory<>(props);
        factory.addListener(new ShareConsumerFactory.Listener<>() {
            @Override
            public void consumerAdded(String id, ShareConsumer<String, String> consumer) {
                log.debug("consumer added id:{}", id);
            }

            @Override
            public void consumerRemoved(@Nullable String id, ShareConsumer<String, String> consumer) {
                log.debug("consumer removed id:{}", id);
            }
        });
        return factory;
    }

    @Bean
    public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerContainerFactory(
            ShareConsumerFactory<String, String> shareConsumerFactory) {
        return new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
    }
}

And used the following test to verify the functionalities.

@Testcontainers
@SpringBootTest
@Slf4j
class DemoApplicationTests {

    // Kafka 4.2 enabled share consumer by default
    @Container
    static KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("apache/kafka:latest"));

    @DynamicPropertySource
    static void kafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafkaContainer::getBootstrapServers);
    }

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private GreetingListener listener;

    @Test
    public void testSendMessage() {
        List.of("the", "quick", "brown", "fox", "jumps", "over", "the", "lazy", "dog")
                .forEach(word -> kafkaTemplate.send(DemoApplication.DEMO_TOPIC_NAME, word)
                        .thenAccept(s -> log.debug("sent message: {}", s)));

        Awaitility.waitAtMost(Duration.ofMillis(30_000))
                .untilAsserted(() -> assertThat(this.listener.getWordCount("the")).isEqualTo(2));
    }

}

When running the tests, the words are sent successfully, but not received by the listener.

The GreetingListener is a simple Kafka listener.

@Component
@Slf4j
public class GreetingListener {
    public Map<String, Long> counter = new ConcurrentHashMap<>();

    @KafkaListener(
            topics = DEMO_TOPIC_NAME,
            containerFactory = "shareKafkaListenerContainerFactory",
            groupId = DEMO_GROUP_NAME
    )
    public void onMessage(ConsumerRecord<String, String> record) {
        log.debug("received record: {} at {}", record, LocalDateTime.now());
        counter.compute(record.value(), (s, v) -> v == null ? 1 : v + 1);
    }

    public Long getWordCount(String word) {
        return this.counter.get(word);
    }
}

[lMysticl]

For this single-broker Share Consumer example, configure the broker as well as the client and group.

Pin the broker image instead of using apache/kafka:latest; the Spring Kafka 4.1 guide used here describes Share Consumers against Kafka 4.2. The Java kafka.version dependency does not control the broker image.

The internal __share_group_state topic also needs a viable single-broker configuration:

share.coordinator.state.topic.replication.factor=1
share.coordinator.state.topic.min.isr=1

Configure share.auto.offset.reset=earliest for the share group through the Admin API when the test may publish before the group starts. It is separate from a regular consumer group's auto.offset.reset. That makes the test independent of whether the listener starts before or after the records are sent; an arbitrary five-second delay is timing-dependent.

The default EXPLICIT ack mode is appropriate for the listener shown here; the container acknowledges after a normal listener return.

This consolidates the broker-side detail added in my follow-up. Reference: Spring Kafka Share Consumers.

[hantsy]

Currently, I set up like this https://github.com/hantsy/spring7-sandbox/blob/master/boot-kafka-share-consumer/src/test/java/com/example/demo/DemoApplicationTests.java#L32 and add a delay before sending messages to resolve the problems, check https://stackoverflow.com/questions/79943319/kafka-share-consumer-not-consume-messages

But it is strange that Spring docs does not mention these at all.

[lMysticl]

Thanks, that clears it up. I missed the broker-side part in my first reply. Share groups use Kafka’s internal __share_group_state topic, so with a single broker, both share.coordinator.state.topic.replication.factor and share.coordinator.state.topic.min.isr need to be set to 1.

I’d still pin the image to apache/kafka:4.2.0 instead of using latest. The five-second delay works, but it’s timing-dependent. Setting share.auto.offset.reset=earliest for the group through the Admin API should make the test independent of whether the listener starts before or after the messages are sent.

It would be helpful if the Spring Kafka docs mentioned this single-broker setup.