Wiretap

Kafka queues

Share groups

Share groups (KIP-932) turn a topic into a queue: the broker hands single records to any member and expects an acknowledgement. wiretap tests them with the same publisher, seeds, assertions and readiness calls as classic listeners, and handles the two things that differ: the broker settings a one-node broker needs, and where a new share group starts reading.

What changes with a share group

A classic consumer group assigns whole partitions to members and tracks a committed offset. A share group does neither. The broker locks each record while a member processes it (group.share.record.lock.duration.ms, 30 seconds by default), the member acknowledges it as ACCEPT, RELEASE or REJECT, and a record that is released or times out is delivered again until group.share.delivery.count.limit (5) archives it. Three consequences for tests: there is no delivery order to assert, there is no offset to read consumption progress from, and a brand-new share group starts at the end of the topic unless told otherwise.

The application side

Spring Boot has no auto-configuration for share consumers, so the application declares a ShareConsumerFactory and a ShareKafkaListenerContainerFactory and points each queue listener at the factory. The groupId names a share group; concurrency is how many members this instance contributes.

KafkaConfig.java
@Bean
ShareConsumerFactory<String, OrderEvent> shareConsumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    return new DefaultShareConsumerFactory<>(props);
}

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

@KafkaListener(topics = "orders.jobs", groupId = "order-workers",
        containerFactory = "shareKafkaListenerContainerFactory", concurrency = "3")
void process(OrderEvent event) {
    orderService.process(event);
}

What wiretap does

  • Embedded broker. The one-node broker starts with share.coordinator.state.topic.replication.factor=1 and share.coordinator.state.topic.min.isr=1. Without them the share coordinator cannot create its state topic and share consumers receive nothing, with no error.
  • Readiness. KafkaTestListeners reports share containers with kind() == SHARE and a members() count read from the broker. A share listener is ready when it runs and at least one member holds an assignment for its topics. awaitAllReady() and awaitListenersForTopic(...) handle a registry that mixes both kinds.
  • Bean overrides. With wiretap.test.bean-overrides.enabled=true a DefaultShareConsumerFactory is pointed at the test broker (the hard-coded localhost:9092 above is fine), production security keys are removed, and share group ids get the per-context suffix, on the factory and on @KafkaListener(groupId). See bean overrides.
  • Start offset. wiretap sets share.auto.offset.reset=earliest on every share group it suffixes, through the group configuration, before the containers start. That is what makes seeded fixtures and records published before the first fetch visible to share listeners.
  • Cleanup. Suffixed share groups are deleted between test methods with deleteShareGroups; groups wiretap did not name are left alone.
  • Unchanged. KafkaTestPublisher, @KafkaSeed, KafkaTopicReader and every assertThatTopic(...) assertion read and write topics, not groups.
readiness failure table
id | topics | running | autoStartup | assigned partitions | kind | members
---|--------|---------|-------------|--------------------|------|--------
orderRouter | [orders.in] | true | true | [orders.in-0] | CLASSIC | 0
orderWorkers | [orders.jobs] | true | true | [] | SHARE | 3

Properties

PropertyDefaultDescription
wiretap.test.broker-properties.<kafka.property>
map
the two share coordinator settingsKafka broker properties applied before the embedded broker starts. The broker is a JVM singleton: the first context wins and later contexts with different keys log a WARN. Ignored in passthrough mode.
wiretap.test.bean-overrides.share.auto-offset-reset
earliest | latest | by_duration:<ISO-8601> | none
earliestValue of share.auto.offset.reset written to each suffixed share group before its containers start. none leaves group configuration untouched. Requires bean overrides to be enabled.
src/test/resources/application.properties
wiretap.test.broker-properties.group.share.record.lock.duration.ms=15000
wiretap.test.broker-properties.group.share.delivery.count.limit=3

Writing the test

The shape is the one from the quick start: wait for the listener, publish, assert on the side effect.

OrderWorkersIT.java
@SpringBootTest
@EnableWiretapKafkaBroker(topics = "orders.jobs")
class OrderWorkersIT {

    @Autowired OrderRepository repository;

    @Test
    void everyJobIsProcessedOnce() {
        KafkaTestListeners.awaitListenersForTopic("orders.jobs");

        Set<String> ids = Set.of("j-1", "j-2", "j-3", "j-4");
        ids.forEach(id -> KafkaTestPublisher.send("orders.jobs", id, new OrderEvent(id)));

        await().atMost(Duration.ofSeconds(10)).untilAsserted(() ->
                assertThat(repository.findAllIds()).containsExactlyInAnyOrderElementsOf(ids));
    }
}

Acknowledgement modes

ShareAckMode.EXPLICIT (the default) acknowledges on the listener's behalf: ACCEPT when the method returns, and whatever the ShareConsumerRecordRecoverer decides when it throws (REJECT by default, RELEASING is the built-in alternative). MANUAL hands the listener a ShareAcknowledgment:

OrderWorkers.java
@KafkaListener(topics = "orders.jobs", containerFactory = "manualShareKafkaListenerContainerFactory")
void process(ConsumerRecord<String, OrderEvent> record, ShareAcknowledgment ack) {
    try {
        orderService.process(record.value());
        ack.acknowledge();          // ACCEPT
    } catch (TransientException e) {
        ack.release();              // RELEASE: delivered again, counts toward the delivery limit
    } catch (Exception e) {
        ack.reject();               // REJECT: archived, never delivered again
    }
}

Test outcomes through behaviour rather than through the acknowledgement API: RELEASE is a second delivery, REJECT is the absence of one, and the delivery limit is a record that stops arriving. For any "not delivered again" claim use a held wait, not a sleep:

Java
// RELEASE shows up as a second delivery; REJECT as no second delivery.
await().during(Duration.ofSeconds(3)).atMost(Duration.ofSeconds(5)).untilAsserted(() ->
        assertThat(listener.deliveriesOf("bad-1")).isEqualTo(1));

Mixed applications

Most services that adopt share groups keep classic listeners too, sometimes on the same topic with different group ids. Nothing needs configuring for that: readiness judges each container by its kind, bean overrides adjust both factory types, cleanup deletes both group types by the same suffix rule, and the assertion API does not care who consumed. The one thing to keep separate in your head is what "consumed" means: a classic listener's progress is an offset, a share listener's progress is a side effect.

Passthrough clusters

Against a cluster you already run, wiretap checks once, when the context declares a share factory or container, that the cluster's finalized share.version is at least 1. If it is not, the context fails with the remedy in the message (kafka-features.sh ... upgrade --feature share.version=1, plus the two replication settings on clusters with fewer than three brokers). wiretap never changes cluster features. A new Kafka 4.2 cluster, such as apache/kafka:4.2.2 in Testcontainers, finalizes share.version=1 by default, so the feature step is only for a 4.1 preview cluster or one upgraded in place; the two replication settings are still needed on a single node. On a locked-down cluster that refuses the group configuration write, the start-offset default is logged as a WARN and skipped.

Properties
wiretap.test.mode=passthrough
wiretap.test.passthrough.bootstrap-servers=localhost:9092

# Only when the cluster refuses group configuration changes:
# wiretap.test.bean-overrides.share.auto-offset-reset=none

What does not apply

  • @RetryableTopic, DefaultErrorHandler and DeadLetterPublishingRecoverer. Share containers use the recoverer and the broker's delivery limit; there is no dead-letter topic unless the application publishes one.
  • Batch listeners, topic patterns, explicit partition assignment and message converters (Spring for Apache Kafka limitations).
  • hasReadyListeners is deprecated but still works for share containers; prefer KafkaTestListeners.awaitListenersForTopic.

Troubleshooting

SymptomCauseFix
Share listener receives nothing, no exceptionNo share-state topic: a real cluster with default replication on fewer than three brokers, or wiretap older than 5.2 on the embedded brokerSet the two share.coordinator.state.topic.* settings, or upgrade wiretap
Context fails and the message mentions share.versionPassthrough cluster older than Kafka 4.2, or the feature is not finalizedRun the kafka-features.sh command from the message
Only the last few published records arriveThe group started at latest while records were in flightKeep the earliest default and call awaitListenersForTopic before publishing
awaitAllReady times out with members = 0The share factory points at another broker (hard-coded address, overrides disabled)Enable bean overrides, or read the address from spring.kafka.bootstrap-servers
UnsupportedOperationException: This container doesn't support retrieving its assigned partitionswiretap older than 5.2 with a share listener in the contextUpgrade wiretap