From 5db56e70a049bcf7b587e74543d1262afc289858 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 02:43:36 +0000 Subject: [PATCH] Add share-groups module: Kafka 4.3.1 share groups with Spring for Apache Kafka 4.1.1 Six consumers on two partitions under a consumer group, a share group with the default share.acquire.mode and a share group with record_limit; the three acknowledgement types and a record that is released until Kafka stops redelivering it; a new share group starting at latest; the ordering a share group gives up; and Spring Kafka's share containers in implicit and manual acknowledgement modes. Tests run against a real broker (scripts/broker-up.sh) and write output/. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01Lm655hdgDo5xsSLNomgdsy --- README.md | 5 +- share-groups/.gitignore | 2 + share-groups/README.md | 59 +++++++ share-groups/output/01-work-spread.txt | 29 ++++ share-groups/output/02-ack-types.txt | 13 ++ .../03-share-group-starts-at-latest.txt | 3 + .../output/04-no-ordering-guarantee.txt | 3 + .../output/05-spring-share-listeners.txt | 10 ++ share-groups/pom.xml | 52 +++++++ share-groups/scripts/broker-down.sh | 3 + share-groups/scripts/broker-up.sh | 40 +++++ share-groups/scripts/run-all.sh | 7 + .../ankurm/sharegroups/PaymentListeners.java | 54 +++++++ .../com/ankurm/sharegroups/ShareConfig.java | 53 +++++++ .../sharegroups/ShareGroupsApplication.java | 9 ++ .../com/ankurm/sharegroups/AckTypesTest.java | 75 +++++++++ .../java/com/ankurm/sharegroups/Broker.java | 67 ++++++++ .../ankurm/sharegroups/OffsetResetTest.java | 60 +++++++ .../com/ankurm/sharegroups/OrderingTest.java | 92 +++++++++++ .../sharegroups/SpringShareListenerTest.java | 71 +++++++++ .../com/ankurm/sharegroups/Transcript.java | 30 ++++ .../ankurm/sharegroups/WorkSpreadTest.java | 146 ++++++++++++++++++ .../src/test/resources/logback-test.xml | 8 + 23 files changed, 889 insertions(+), 2 deletions(-) create mode 100644 share-groups/.gitignore create mode 100644 share-groups/README.md create mode 100644 share-groups/output/01-work-spread.txt create mode 100644 share-groups/output/02-ack-types.txt create mode 100644 share-groups/output/03-share-group-starts-at-latest.txt create mode 100644 share-groups/output/04-no-ordering-guarantee.txt create mode 100644 share-groups/output/05-spring-share-listeners.txt create mode 100644 share-groups/pom.xml create mode 100755 share-groups/scripts/broker-down.sh create mode 100755 share-groups/scripts/broker-up.sh create mode 100755 share-groups/scripts/run-all.sh create mode 100644 share-groups/src/main/java/com/ankurm/sharegroups/PaymentListeners.java create mode 100644 share-groups/src/main/java/com/ankurm/sharegroups/ShareConfig.java create mode 100644 share-groups/src/main/java/com/ankurm/sharegroups/ShareGroupsApplication.java create mode 100644 share-groups/src/test/java/com/ankurm/sharegroups/AckTypesTest.java create mode 100644 share-groups/src/test/java/com/ankurm/sharegroups/Broker.java create mode 100644 share-groups/src/test/java/com/ankurm/sharegroups/OffsetResetTest.java create mode 100644 share-groups/src/test/java/com/ankurm/sharegroups/OrderingTest.java create mode 100644 share-groups/src/test/java/com/ankurm/sharegroups/SpringShareListenerTest.java create mode 100644 share-groups/src/test/java/com/ankurm/sharegroups/Transcript.java create mode 100644 share-groups/src/test/java/com/ankurm/sharegroups/WorkSpreadTest.java create mode 100644 share-groups/src/test/resources/logback-test.xml diff --git a/README.md b/README.md index 4d24147..9b9f8bc 100644 --- a/README.md +++ b/README.md @@ -3,8 +3,8 @@ Companion code for the messaging series on [ankurm.com](https://ankurm.com). Each directory is a self-contained Maven project for one article, with its own `pom.xml` and its own captured output, regenerated from the test suite rather than typed by hand. Most modules keep that -output under `docs/output/` alongside numbered documentation chapters; `saga/` keeps it directly -under `output/` instead, with the equivalent depth in the article's own accordion sections. +output under `docs/output/` alongside numbered documentation chapters; `saga/` and `share-groups/` keep it +directly under `output/` instead, with the equivalent depth in the article's own accordion sections. | Module | Article | What it demonstrates | |---|---|---| @@ -15,6 +15,7 @@ under `output/` instead, with the equivalent depth in the article's own accordio | [`protocol-comparison/`](protocol-comparison/README.md) | [RSocket vs gRPC vs WebSocket on Spring Boot 4.1: When Each One Wins](https://ankurm.com/rsocket-vs-grpc-vs-websocket-spring-boot-4-1/) | One application serving the same two operations over all three, benchmarked in one JVM, ending in the only measurement that transfers off the box: what each server produces for a consumer that asked for a hundred | | [`broker-comparison/`](broker-comparison/README.md) | [Kafka vs RabbitMQ vs Pulsar for Java Teams: A Decision Framework with Benchmarks](https://ankurm.com/kafka-vs-rabbitmq-vs-pulsar-java-decision-framework/) | The three brokers measured side by side on ordering, replay, consumer scaling and operational footprint, ending in a decision table where every row has a transcript behind it | | [`saga/`](saga/README.md) | Saga Pattern in Spring Boot: Orchestration vs Choreography with Kafka | The same order/payment/inventory saga built twice against a real broker — choreography and orchestration — driven through the same inventory failure so the compensating transaction can be compared side by side, plus a real duplicate-delivery test for the idempotent-consumer problem | +| [`share-groups/`](share-groups/README.md) | [Queues for Kafka: Share Groups with Spring for Apache Kafka 4](https://ankurm.com/kafka-share-groups-spring-kafka-4-queues-for-kafka/) | Share groups against a real Kafka 4.3.1 broker: six consumers on two partitions, the acquire mode that decides whether they really share, the three acknowledgement types, a group that starts at the end of the topic, and the ordering you give up | The brokers make an instructive set. Kafka's consumer holds an offset and the broker remembers nothing about individual records; RabbitMQ's broker owns the message until it is acknowledged and diff --git a/share-groups/.gitignore b/share-groups/.gitignore new file mode 100644 index 0000000..a7b0fe2 --- /dev/null +++ b/share-groups/.gitignore @@ -0,0 +1,2 @@ +target/ +.kafka/ diff --git a/share-groups/README.md b/share-groups/README.md new file mode 100644 index 0000000..a8eeb5a --- /dev/null +++ b/share-groups/README.md @@ -0,0 +1,59 @@ +# share-groups + +Companion code for *Queues for Kafka: Share Groups with Spring for Apache Kafka 4* on +[ankurm.com](https://ankurm.com/kafka-share-groups-spring-kafka-4-queues-for-kafka/). + +What a Kafka **share group** (KIP-932) does differently from a consumer group, measured against a +real broker: how many consumers can share one partition, what the three acknowledgement types do to a +record, why a new share group sees nothing, and what ordering you give up. The same ideas are then run +through Spring Kafka's share containers. + +## Versions + +| Artifact | Version | +|---|---| +| Spring Boot | 4.1.1 (GA; `4.2.0-M2` is a milestone) | +| Spring Kafka | 4.1.1, managed by the Boot BOM (share consumer support is already in this GA release) | +| kafka-clients | 4.2.1, managed by the Boot BOM | +| Kafka broker | 4.3.1, started by `scripts/broker-up.sh` (`share.version` is finalized at level 1 by default; no feature flag was set) | +| JDK | 25 (LTS) | + +## Quickstart + +``` +scripts/broker-up.sh # downloads Kafka 4.3.1 (checksum-verified), starts KRaft on localhost:9092 +scripts/run-all.sh # runs every test and rewrites output/*.txt +scripts/broker-down.sh +``` + +The tests do not start a broker themselves. Use `-Dbroker=host:port` if yours is elsewhere. Each test +creates its own uniquely named topic and group, so reruns never see earlier records. + +## What each test is + +| Test | Role | +|---|---| +| `WorkSpreadTest` | 6 consumers, a 2-partition topic, 60 records at 50 ms each: consumer group, share group with the default `share.acquire.mode`, and share group with `record_limit` | +| `AckTypesTest` | ACCEPT, RELEASE and REJECT through the raw `KafkaShareConsumer`, including a record that is released every time | +| `OffsetResetTest` | A new share group starts at the latest offset; `share.auto.offset.reset=earliest` is a group config set through the Admin API | +| `OrderingTest` | One key, one partition, three consumers of different speed: arrival order is not produce order | +| `SpringShareListenerTest` | `ShareKafkaListenerContainerFactory` in implicit and manual acknowledgement modes with `@KafkaListener` | + +`ShareConfig` is the hand-written configuration: Spring Boot 4.1.1 has no auto-configuration for share +consumers. + +## Captured output (`output/`) + +| File | What it captures | +|---|---| +| `01-work-spread.txt` | Which of six consumers received records under each of the three setups, with indicative timings | +| `02-ack-types.txt` | Every delivery of four records: accepted, released once, rejected, released until Kafka stops redelivering | +| `03-share-group-starts-at-latest.txt` | 3 records produced first: 0 received by default, 3 received after switching the group to earliest | +| `04-no-ordering-guarantee.txt` | The order 40 same-key records were processed in, and the number of out-of-order neighbours | +| `05-spring-share-listeners.txt` | Listener threads under implicit mode, and per-record release, reject and acknowledge under manual mode | + +Timings and the exact split of records between consumers differ from run to run. The shape of each +result is stable; the milliseconds are not. + +No `docs/` chapter directory in this repository -- the intermediate and reference-depth material is in +accordion sections inside the WordPress post. diff --git a/share-groups/output/01-work-spread.txt b/share-groups/output/01-work-spread.txt new file mode 100644 index 0000000..b6730e9 --- /dev/null +++ b/share-groups/output/01-work-spread.txt @@ -0,0 +1,29 @@ +topic partitions = 2, consumers = 6, records = 60, work per record = 50 ms +== consumer group (6 KafkaConsumers, same group.id) +consumer-0 processed 26 records +consumer-1 processed 34 records +consumer-2 processed 0 records +consumer-3 processed 0 records +consumer-4 processed 0 records +consumer-5 processed 0 records +total processed = 60 in 1993 ms +== share group (6 ShareConsumers, default share.acquire.mode) +consumer-0 processed 26 records +consumer-1 processed 34 records +consumer-2 processed 0 records +consumer-3 processed 0 records +consumer-4 processed 0 records +consumer-5 processed 0 records +total processed = 60 in 1778 ms +== share group (6 ShareConsumers, share.acquire.mode=record_limit) +consumer-0 processed 9 records +consumer-1 processed 12 records +consumer-2 processed 10 records +consumer-3 processed 8 records +consumer-4 processed 9 records +consumer-5 processed 12 records +total processed = 60 in 681 ms +consumer group : consumers that received records = 2 +share group, default share.acquire.mode: consumers that received records = 2 +share group, acquire.mode=record_limit : consumers that received records = 6 +(elapsed times vary from run to run; the shape, not the milliseconds, is the result) diff --git a/share-groups/output/02-ack-types.txt b/share-groups/output/02-ack-types.txt new file mode 100644 index 0000000..a965347 --- /dev/null +++ b/share-groups/output/02-ack-types.txt @@ -0,0 +1,13 @@ +ok delivery #1 -> accept +retry-once delivery #1 -> release +reject-me delivery #1 -> reject +always-release delivery #1 -> release +retry-once delivery #2 -> accept +always-release delivery #2 -> release +always-release delivery #3 -> release +always-release delivery #4 -> release +always-release delivery #5 -> release +deliveries of ok = 1 +deliveries of retry-once = 2 +deliveries of reject-me = 1 +deliveries of always-release = 5 diff --git a/share-groups/output/03-share-group-starts-at-latest.txt b/share-groups/output/03-share-group-starts-at-latest.txt new file mode 100644 index 0000000..4f527bb --- /dev/null +++ b/share-groups/output/03-share-group-starts-at-latest.txt @@ -0,0 +1,3 @@ +produced 3 records before any group existed +new group, default share.auto.offset.reset -> received 0 records +new group, share.auto.offset.reset=earliest -> received 3 records diff --git a/share-groups/output/04-no-ordering-guarantee.txt b/share-groups/output/04-no-ordering-guarantee.txt new file mode 100644 index 0000000..b331fea --- /dev/null +++ b/share-groups/output/04-no-ordering-guarantee.txt @@ -0,0 +1,3 @@ +produced : 0..39 in order, one key, one partition +processed : [4, 5, 6, 7, 12, 0, 13, 14, 15, 16, 17, 18, 19, 1, 20, 8, 21, 22, 23, 24, 2, 25, 26, 27, 28, 29, 30, 3, 31, 36, 9, 37, 38, 39, 32, 33, 10, 34, 35, 11] +adjacent pairs that arrived out of order = 9 diff --git a/share-groups/output/05-spring-share-listeners.txt b/share-groups/output/05-spring-share-listeners.txt new file mode 100644 index 0000000..b370afb --- /dev/null +++ b/share-groups/output/05-spring-share-listeners.txt @@ -0,0 +1,10 @@ +== implicit mode, concurrency 4, 24 records on one topic partition +records handled = 24 +distinct listener threads that handled records = 2 + implicit-C-1 + implicit-C-2 +== manual mode (ShareAckMode.MANUAL), listener takes a ShareAcknowledgment +again delivery #1 -> release +again delivery #2 -> acknowledge +bad delivery #1 -> reject +ok delivery #1 -> acknowledge diff --git a/share-groups/pom.xml b/share-groups/pom.xml new file mode 100644 index 0000000..32d0134 --- /dev/null +++ b/share-groups/pom.xml @@ -0,0 +1,52 @@ + + 4.0.0 + + + + org.springframework.boot + spring-boot-starter-parent + 4.1.1 + + + + com.ankurm + share-groups + 1.0 + jar + + + 25 + UTF-8 + + + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-kafka + + + org.springframework.boot + spring-boot-starter-test + test + + + org.awaitility + awaitility + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/share-groups/scripts/broker-down.sh b/share-groups/scripts/broker-down.sh new file mode 100755 index 0000000..d469ba3 --- /dev/null +++ b/share-groups/scripts/broker-down.sh @@ -0,0 +1,3 @@ +#!/usr/bin/env bash +# Stops the broker started by broker-up.sh (matched by its main class, not by a broad pattern). +for p in $(ps -eo pid,cmd | grep '[k]afka.Kafka' | awk '{print $1}'); do kill "$p"; done diff --git a/share-groups/scripts/broker-up.sh b/share-groups/scripts/broker-up.sh new file mode 100755 index 0000000..a3b6dd0 --- /dev/null +++ b/share-groups/scripts/broker-up.sh @@ -0,0 +1,40 @@ +#!/usr/bin/env bash +# Downloads Apache Kafka 4.3.1 (checksum-verified), formats a single-node KRaft cluster and starts it. +# Usage: scripts/broker-up.sh [client-port] (default 9092; the controller uses client-port + 1) +set -euo pipefail + +VERSION=4.3.1 +PORT="${1:-9092}" +CTRL=$((PORT + 1)) +HOME_DIR="${KAFKA_DIR:-$(cd "$(dirname "$0")/.." && pwd)/.kafka}" +TGZ="kafka_2.13-${VERSION}.tgz" +mkdir -p "$HOME_DIR" && cd "$HOME_DIR" + +if [ ! -d "kafka_2.13-${VERSION}" ]; then + curl -fsSL -o "$TGZ" "https://downloads.apache.org/kafka/${VERSION}/${TGZ}" + curl -fsSL -o "${TGZ}.sha512" "https://downloads.apache.org/kafka/${VERSION}/${TGZ}.sha512" + # The .sha512 file is "name: HEX HEX HEX ..." in groups, so compare after stripping spaces and the name. + expected="$(sed 's/^[^:]*: *//' "${TGZ}.sha512" | tr -d ' \n' | tr 'A-F' 'a-f')" + actual="$(sha512sum "$TGZ" | cut -d' ' -f1)" + [ "$expected" = "$actual" ] || { echo "checksum mismatch for $TGZ" >&2; exit 1; } + tar -xzf "$TGZ" +fi + +cd "kafka_2.13-${VERSION}" +sed -i "s#^listeners=.*#listeners=PLAINTEXT://:${PORT},CONTROLLER://:${CTRL}#" config/server.properties +sed -i "s#^advertised.listeners=.*#advertised.listeners=PLAINTEXT://localhost:${PORT}#" config/server.properties +sed -i "s#^controller.quorum.bootstrap.servers=.*#controller.quorum.bootstrap.servers=localhost:${CTRL}#" config/server.properties +sed -i "s#^log.dirs=.*#log.dirs=${HOME_DIR}/data-${PORT}#" config/server.properties + +if [ ! -d "${HOME_DIR}/data-${PORT}" ]; then + bin/kafka-storage.sh format --standalone -t "$(bin/kafka-storage.sh random-uuid | tail -1)" -c config/server.properties +fi +export KAFKA_HEAP_OPTS="${KAFKA_HEAP_OPTS:--Xmx512m -Xms512m}" +setsid nohup bin/kafka-server-start.sh config/server.properties > "${HOME_DIR}/broker-${PORT}.log" 2>&1 < /dev/null & + +for _ in $(seq 1 60); do + bin/kafka-topics.sh --bootstrap-server "localhost:${PORT}" --list > /dev/null 2>&1 && break + sleep 2 +done +echo "broker up on localhost:${PORT}; share.version feature:" +bin/kafka-features.sh --bootstrap-server "localhost:${PORT}" describe | grep share.version diff --git a/share-groups/scripts/run-all.sh b/share-groups/scripts/run-all.sh new file mode 100755 index 0000000..9dbfd1e --- /dev/null +++ b/share-groups/scripts/run-all.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +# Regenerates every file in output/ from the tests. Needs a broker: scripts/broker-up.sh [port], then +# pass -Dbroker=localhost: if it is not 9092. +set -euo pipefail +cd "$(dirname "$0")/.." +mvn -B -q test "$@" +ls -1 output diff --git a/share-groups/src/main/java/com/ankurm/sharegroups/PaymentListeners.java b/share-groups/src/main/java/com/ankurm/sharegroups/PaymentListeners.java new file mode 100644 index 0000000..a2eb781 --- /dev/null +++ b/share-groups/src/main/java/com/ankurm/sharegroups/PaymentListeners.java @@ -0,0 +1,54 @@ +package com.ankurm.sharegroups; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.kafka.clients.consumer.ConsumerRecord; + +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.support.ShareAcknowledgment; +import org.springframework.stereotype.Component; + +/** Two listeners on share containers. They only record what they saw; the tests read the lists. */ +@Component +public class PaymentListeners { + + public final List implicitSeen = new CopyOnWriteArrayList<>(); + + public final List manualSeen = new CopyOnWriteArrayList<>(); + + /** Implicit mode: returning normally acknowledges (ACCEPT); throwing makes the container reject or release. */ + @KafkaListener(id = "implicit", topics = "${share.topic.implicit}", groupId = "${share.group.implicit}", + containerFactory = "implicitShareFactory") + void implicit(ConsumerRecord record) throws InterruptedException { + Thread.sleep(20); + implicitSeen.add(Thread.currentThread().getName() + " <- " + record.value()); + } + + /** Manual mode: the listener decides per record. "bad" is rejected, "again" is released once, the rest accepted. */ + @KafkaListener(id = "manual", topics = "${share.topic.manual}", groupId = "${share.group.manual}", + containerFactory = "explicitShareFactory") + void manual(ConsumerRecord record, ShareAcknowledgment ack) { + int attempt = record.deliveryCount().orElse((short) 1); + switch (record.value()) { + case "bad" -> { + manualSeen.add("bad delivery #" + attempt + " -> reject"); + ack.reject(); + } + case "again" -> { + if (attempt == 1) { + manualSeen.add("again delivery #1 -> release"); + ack.release(); + } + else { + manualSeen.add("again delivery #" + attempt + " -> acknowledge"); + ack.acknowledge(); + } + } + default -> { + manualSeen.add(record.value() + " delivery #" + attempt + " -> acknowledge"); + ack.acknowledge(); + } + } + } +} diff --git a/share-groups/src/main/java/com/ankurm/sharegroups/ShareConfig.java b/share-groups/src/main/java/com/ankurm/sharegroups/ShareConfig.java new file mode 100644 index 0000000..f9efef3 --- /dev/null +++ b/share-groups/src/main/java/com/ankurm/sharegroups/ShareConfig.java @@ -0,0 +1,53 @@ +package com.ankurm.sharegroups; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.config.ShareKafkaListenerContainerFactory; +import org.springframework.kafka.core.DefaultShareConsumerFactory; +import org.springframework.kafka.core.ShareConsumerFactory; +import org.springframework.kafka.listener.ContainerProperties.ShareAckMode; + +/** + * Spring Boot 4.1.1 does not auto-configure share consumers, so the factory and the listener container + * factories are plain beans. Two container factories: implicit (the container acknowledges for you) and + * explicit (the listener must call ShareAcknowledgment). + */ +@Configuration +@EnableKafka +public class ShareConfig { + + @Bean + ShareConsumerFactory shareConsumerFactory(@Value("${broker:localhost:9092}") String bootstrap) { + Map props = new HashMap<>(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + props.put("share.acquire.mode", "record_limit"); + props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "3"); + return new DefaultShareConsumerFactory<>(props); + } + + @Bean + ShareKafkaListenerContainerFactory implicitShareFactory(ShareConsumerFactory factory) { + ShareKafkaListenerContainerFactory f = new ShareKafkaListenerContainerFactory<>(factory); + f.setConcurrency(4); + f.getContainerProperties().setShareAckMode(ShareAckMode.IMPLICIT); + return f; + } + + @Bean + ShareKafkaListenerContainerFactory explicitShareFactory(ShareConsumerFactory factory) { + ShareKafkaListenerContainerFactory f = new ShareKafkaListenerContainerFactory<>(factory); + f.setConcurrency(2); + f.getContainerProperties().setShareAckMode(ShareAckMode.MANUAL); + return f; + } +} diff --git a/share-groups/src/main/java/com/ankurm/sharegroups/ShareGroupsApplication.java b/share-groups/src/main/java/com/ankurm/sharegroups/ShareGroupsApplication.java new file mode 100644 index 0000000..75ed51d --- /dev/null +++ b/share-groups/src/main/java/com/ankurm/sharegroups/ShareGroupsApplication.java @@ -0,0 +1,9 @@ +package com.ankurm.sharegroups; + +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** Marker application class so the Spring tests have something to bootstrap. Nothing here starts a broker. */ +@SpringBootApplication +public class ShareGroupsApplication { + +} diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/AckTypesTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/AckTypesTest.java new file mode 100644 index 0000000..d02b8e2 --- /dev/null +++ b/share-groups/src/test/java/com/ankurm/sharegroups/AckTypesTest.java @@ -0,0 +1,75 @@ +package com.ankurm.sharegroups; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +import org.apache.kafka.clients.consumer.AcknowledgeType; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.KafkaShareConsumer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** What ACCEPT, RELEASE and REJECT do to a record, and what happens to a record that is released forever. */ +class AckTypesTest { + + @Test + void acknowledgementTypes() throws Exception { + Transcript t = new Transcript("02-ack-types.txt"); + String topic = Broker.freshTopic("acks", 1); + String group = topic + "-g"; + Broker.shareGroupFromEarliest(group); + Broker.produce(topic, List.of(new String[] { "k", "ok" }, new String[] { "k", "retry-once" }, + new String[] { "k", "reject-me" }, new String[] { "k", "always-release" })); + + Properties p = new Properties(); + p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Broker.BOOTSTRAP); + p.put(ConsumerConfig.GROUP_ID_CONFIG, group); + p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + p.put("share.acknowledgement.mode", "explicit"); + + List deliveries = new ArrayList<>(); + long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(40); + long lastRecord = System.nanoTime(); + try (KafkaShareConsumer c = new KafkaShareConsumer<>(p)) { + c.subscribe(List.of(topic)); + while (System.nanoTime() < end && System.nanoTime() - lastRecord < TimeUnit.SECONDS.toNanos(15)) { + for (ConsumerRecord r : c.poll(Duration.ofMillis(500))) { + lastRecord = System.nanoTime(); + int attempt = r.deliveryCount().orElse((short) -1); + AcknowledgeType type = switch (r.value()) { + case "ok" -> AcknowledgeType.ACCEPT; + case "retry-once" -> attempt == 1 ? AcknowledgeType.RELEASE : AcknowledgeType.ACCEPT; + case "reject-me" -> AcknowledgeType.REJECT; + default -> AcknowledgeType.RELEASE; + }; + c.acknowledge(r, type); + String line = String.format("%-15s delivery #%d -> %s", r.value(), attempt, type); + deliveries.add(line); + t.line("%s", line); + } + c.commitSync(); + } + } + t.line("deliveries of ok = %d", count(deliveries, "ok ")); + t.line("deliveries of retry-once = %d", count(deliveries, "retry-once")); + t.line("deliveries of reject-me = %d", count(deliveries, "reject-me")); + t.line("deliveries of always-release = %d", count(deliveries, "always-release")); + t.write(); + + assertThat(count(deliveries, "ok ")).isEqualTo(1); + assertThat(count(deliveries, "retry-once")).isEqualTo(2); + assertThat(count(deliveries, "reject-me")).isEqualTo(1); + assertThat(count(deliveries, "always-release")).isGreaterThan(1); + } + + private long count(List lines, String prefix) { + return lines.stream().filter(l -> l.startsWith(prefix)).count(); + } +} diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/Broker.java b/share-groups/src/test/java/com/ankurm/sharegroups/Broker.java new file mode 100644 index 0000000..204d7d4 --- /dev/null +++ b/share-groups/src/test/java/com/ankurm/sharegroups/Broker.java @@ -0,0 +1,67 @@ +package com.ankurm.sharegroups; + +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.AlterConfigOp; +import org.apache.kafka.clients.admin.ConfigEntry; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.config.ConfigResource; +import org.apache.kafka.common.serialization.StringSerializer; + +/** + * Helpers for the real broker. The tests do not start a broker: run scripts/broker-up.sh first + * (Kafka 4.3.1, KRaft, localhost:9092). Override with -Dbroker=host:port. + */ +final class Broker { + + static final String BOOTSTRAP = System.getProperty("broker", "localhost:9092"); + + private Broker() { + } + + static Properties admin() { + Properties p = new Properties(); + p.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP); + return p; + } + + /** A fresh topic per test run (unique suffix), so reruns never see old records. */ + static String freshTopic(String base, int partitions) throws Exception { + String name = base + "-" + System.currentTimeMillis(); + try (AdminClient admin = AdminClient.create(admin())) { + admin.createTopics(List.of(new NewTopic(name, partitions, (short) 1))).all().get(30, TimeUnit.SECONDS); + } + return name; + } + + /** Share groups start at the latest offset by default; flip a group to earliest before it first subscribes. */ + static void shareGroupFromEarliest(String group) throws Exception { + ConfigResource resource = new ConfigResource(ConfigResource.Type.GROUP, group); + try (AdminClient admin = AdminClient.create(admin())) { + admin.incrementalAlterConfigs(Map.of(resource, + List.of(new AlterConfigOp(new ConfigEntry("share.auto.offset.reset", "earliest"), AlterConfigOp.OpType.SET)))) + .all().get(30, TimeUnit.SECONDS); + } + } + + static void produce(String topic, List keyValues) { + Properties p = new Properties(); + p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP); + p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + try (KafkaProducer producer = new KafkaProducer<>(p)) { + for (String[] kv : keyValues) { + producer.send(new ProducerRecord<>(topic, kv[0], kv[1])); + } + producer.flush(); + } + } +} diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/OffsetResetTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/OffsetResetTest.java new file mode 100644 index 0000000..6d72561 --- /dev/null +++ b/share-groups/src/test/java/com/ankurm/sharegroups/OffsetResetTest.java @@ -0,0 +1,60 @@ +package com.ankurm.sharegroups; + +import java.time.Duration; +import java.util.List; +import java.util.Properties; + +import org.apache.kafka.clients.consumer.AcknowledgeType; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.KafkaShareConsumer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** The first surprise: a brand-new share group starts at the end of the topic, not the beginning. */ +class OffsetResetTest { + + @Test + void newShareGroupSeesNothingUntilItIsToldToStartEarly() throws Exception { + Transcript t = new Transcript("03-share-group-starts-at-latest.txt"); + String topic = Broker.freshTopic("reset", 1); + Broker.produce(topic, List.of(new String[] { "k", "a" }, new String[] { "k", "b" }, new String[] { "k", "c" })); + t.line("produced 3 records before any group existed"); + + int byDefault = drain(topic, topic + "-default", t); + t.line("new group, default share.auto.offset.reset -> received %d records", byDefault); + + String early = topic + "-early"; + Broker.shareGroupFromEarliest(early); + int withEarliest = drain(topic, early, t); + t.line("new group, share.auto.offset.reset=earliest -> received %d records", withEarliest); + t.write(); + + assertThat(byDefault).isZero(); + assertThat(withEarliest).isEqualTo(3); + } + + private int drain(String topic, String group, Transcript t) { + Properties p = new Properties(); + p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Broker.BOOTSTRAP); + p.put(ConsumerConfig.GROUP_ID_CONFIG, group); + p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + p.put("share.acknowledgement.mode", "explicit"); + int received = 0; + try (KafkaShareConsumer c = new KafkaShareConsumer<>(p)) { + c.subscribe(List.of(topic)); + long end = System.currentTimeMillis() + 12_000; + while (System.currentTimeMillis() < end) { + for (ConsumerRecord r : c.poll(Duration.ofMillis(500))) { + c.acknowledge(r, AcknowledgeType.ACCEPT); + received++; + } + c.commitSync(); + } + } + return received; + } +} diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/OrderingTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/OrderingTest.java new file mode 100644 index 0000000..3d9eea7 --- /dev/null +++ b/share-groups/src/test/java/com/ankurm/sharegroups/OrderingTest.java @@ -0,0 +1,92 @@ +package com.ankurm.sharegroups; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.kafka.clients.consumer.AcknowledgeType; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.KafkaShareConsumer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** 40 records, one key, one partition, three consumers with uneven speed: arrival order is not the produce order. */ +class OrderingTest { + + @Test + void sameKeyDoesNotMeanSameOrder() throws Exception { + Transcript t = new Transcript("04-no-ordering-guarantee.txt"); + String topic = Broker.freshTopic("order", 1); + String group = topic + "-g"; + Broker.shareGroupFromEarliest(group); + List records = new ArrayList<>(); + for (int i = 0; i < 40; i++) { + records.add(new String[] { "same-key", String.valueOf(i) }); + } + + List arrival = Collections.synchronizedList(new ArrayList<>()); + AtomicInteger done = new AtomicInteger(); + ExecutorService pool = Executors.newFixedThreadPool(3); + int[] workMillis = { 5, 40, 90 }; + for (int id = 0; id < 3; id++) { + int work = workMillis[id]; + pool.submit(() -> { + Properties p = new Properties(); + p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Broker.BOOTSTRAP); + p.put(ConsumerConfig.GROUP_ID_CONFIG, group); + p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "4"); + p.put("share.acknowledgement.mode", "explicit"); + p.put("share.acquire.mode", "record_limit"); + try (KafkaShareConsumer c = new KafkaShareConsumer<>(p)) { + c.subscribe(List.of(topic)); + while (done.get() < 40 && !Thread.currentThread().isInterrupted()) { + for (ConsumerRecord r : c.poll(Duration.ofMillis(300))) { + Thread.sleep(work); + arrival.add(Integer.valueOf(r.value())); + c.acknowledge(r, AcknowledgeType.ACCEPT); + done.incrementAndGet(); + } + c.commitSync(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + } + Thread.sleep(6000); // let all three consumers join the group before any record exists + Broker.produce(topic, records); + long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(60); + while (done.get() < 40 && System.nanoTime() < end) { + Thread.sleep(100); + } + pool.shutdownNow(); + pool.awaitTermination(15, TimeUnit.SECONDS); + + List snapshot = new ArrayList<>(arrival); + long inversions = 0; + for (int i = 1; i < snapshot.size(); i++) { + if (snapshot.get(i) < snapshot.get(i - 1)) { + inversions++; + } + } + t.line("produced : 0..39 in order, one key, one partition"); + t.line("processed : %s", snapshot); + t.line("adjacent pairs that arrived out of order = %d", inversions); + t.write(); + + assertThat(snapshot).hasSize(40); + assertThat(inversions).isGreaterThan(0); + } +} diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/SpringShareListenerTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/SpringShareListenerTest.java new file mode 100644 index 0000000..9047c99 --- /dev/null +++ b/share-groups/src/test/java/com/ankurm/sharegroups/SpringShareListenerTest.java @@ -0,0 +1,71 @@ +package com.ankurm.sharegroups; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.TreeSet; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** The same ideas through Spring Kafka 4.1.1's share containers instead of the raw client. */ +@SpringBootTest(classes = { ShareGroupsApplication.class }) +class SpringShareListenerTest { + + static String implicitTopic; + static String manualTopic; + + @DynamicPropertySource + static void props(DynamicPropertyRegistry r) throws Exception { + implicitTopic = Broker.freshTopic("spring-implicit", 1); + manualTopic = Broker.freshTopic("spring-manual", 1); + Broker.shareGroupFromEarliest(implicitTopic + "-g"); + Broker.shareGroupFromEarliest(manualTopic + "-g"); + r.add("share.topic.implicit", () -> implicitTopic); + r.add("share.group.implicit", () -> implicitTopic + "-g"); + r.add("share.topic.manual", () -> manualTopic); + r.add("share.group.manual", () -> manualTopic + "-g"); + r.add("spring.kafka.listener.auto-startup", () -> "true"); + } + + @Autowired + PaymentListeners listeners; + + @Test + void implicitAndManualAcknowledgement() throws Exception { + Transcript t = new Transcript("05-spring-share-listeners.txt"); + + List batch = new ArrayList<>(); + for (int i = 0; i < 24; i++) { + batch.add(new String[] { "k" + i, "p" + i }); + } + Broker.produce(implicitTopic, batch); + await().atMost(Duration.ofSeconds(60)).until(() -> listeners.implicitSeen.size() >= 24); + + Set threads = new TreeSet<>(); + listeners.implicitSeen.forEach(s -> threads.add(s.substring(0, s.indexOf(" <- ")))); + t.line("== implicit mode, concurrency 4, 24 records on one topic partition"); + t.line("records handled = %d", listeners.implicitSeen.size()); + t.line("distinct listener threads that handled records = %d", threads.size()); + threads.forEach(n -> t.line(" %s", n)); + + Broker.produce(manualTopic, List.of(new String[] { "k", "ok" }, new String[] { "k", "again" }, new String[] { "k", "bad" })); + await().atMost(Duration.ofSeconds(60)).until(() -> listeners.manualSeen.size() >= 4); + Thread.sleep(8000); + t.line("== manual mode (ShareAckMode.MANUAL), listener takes a ShareAcknowledgment"); + listeners.manualSeen.stream().sorted().forEach(s -> t.line("%s", s)); + t.write(); + + assertThat(listeners.implicitSeen).hasSize(24); + assertThat(listeners.manualSeen).anyMatch(s -> s.startsWith("again delivery #2")); + assertThat(listeners.manualSeen.stream().filter(s -> s.startsWith("bad")).count()).isEqualTo(1); + } +} diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/Transcript.java b/share-groups/src/test/java/com/ankurm/sharegroups/Transcript.java new file mode 100644 index 0000000..642518b --- /dev/null +++ b/share-groups/src/test/java/com/ankurm/sharegroups/Transcript.java @@ -0,0 +1,30 @@ +package com.ankurm.sharegroups; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +/** Collects the lines a test prints and writes them to output/NN-name.txt, so the article quotes real output. */ +final class Transcript { + + private final String file; + private final List lines = new ArrayList<>(); + + Transcript(String file) { + this.file = file; + } + + synchronized void line(String format, Object... args) { + String l = String.format(format, args); + lines.add(l); + System.out.println(l); + } + + void write() throws IOException { + Path dir = Path.of("output"); + Files.createDirectories(dir); + Files.write(dir.resolve(file), lines); + } +} diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/WorkSpreadTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/WorkSpreadTest.java new file mode 100644 index 0000000..9b8dc77 --- /dev/null +++ b/share-groups/src/test/java/com/ankurm/sharegroups/WorkSpreadTest.java @@ -0,0 +1,146 @@ +package com.ankurm.sharegroups; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.kafka.clients.consumer.AcknowledgeType; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.KafkaShareConsumer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Six consumers, a two-partition topic, 60 records that each take 50 ms to process. + * A consumer group can keep at most two of the six busy; a share group can keep more busy, but only when + * share.acquire.mode lets max.poll.records bound what one consumer takes (see the default-mode result). + */ +class WorkSpreadTest { + + static final int CONSUMERS = 6; + static final int RECORDS = 60; + static final int PARTITIONS = 2; + + record Result(ConcurrentHashMap perConsumer, long millis) { + } + + @Test + void consumerGroupVersusShareGroup() throws Exception { + Transcript t = new Transcript("01-work-spread.txt"); + t.line("topic partitions = %d, consumers = %d, records = %d, work per record = 50 ms", PARTITIONS, CONSUMERS, RECORDS); + + Result classic = run("classic", t); + Result shareDefault = run("share-default", t); + Result share = run("share-record-limit", t); + + long activeClassic = classic.perConsumer().values().stream().filter(c -> c.get() > 0).count(); + long activeShare = share.perConsumer().values().stream().filter(c -> c.get() > 0).count(); + long activeDefault = shareDefault.perConsumer().values().stream().filter(c -> c.get() > 0).count(); + t.line("consumer group : consumers that received records = %d", activeClassic); + t.line("share group, default share.acquire.mode: consumers that received records = %d", activeDefault); + t.line("share group, acquire.mode=record_limit : consumers that received records = %d", activeShare); + t.line("(elapsed times vary from run to run; the shape, not the milliseconds, is the result)"); + t.write(); + + assertThat(activeClassic).isEqualTo(PARTITIONS); + assertThat(activeShare).isGreaterThan(PARTITIONS); + assertThat(activeDefault).isGreaterThanOrEqualTo(1); + } + + private Result run(String variant, Transcript t) throws Exception { + boolean share = variant.startsWith("share"); + String topic = Broker.freshTopic("spread-" + variant, PARTITIONS); + String group = topic + "-g"; + if (share) { + Broker.shareGroupFromEarliest(group); + } + ConcurrentHashMap counts = new ConcurrentHashMap<>(); + AtomicInteger total = new AtomicInteger(); + ExecutorService pool = Executors.newFixedThreadPool(CONSUMERS); + for (int i = 0; i < CONSUMERS; i++) { + int id = i; + counts.put(id, new AtomicInteger()); + pool.submit(() -> consume(variant, topic, group, id, counts, total)); + } + Thread.sleep(6000); // let every consumer join before the records arrive + List records = new ArrayList<>(); + for (int i = 0; i < RECORDS; i++) { + records.add(new String[] { "key-" + i, "r" + i }); + } + long start = System.nanoTime(); + Broker.produce(topic, records); + while (total.get() < RECORDS && System.nanoTime() - start < TimeUnit.SECONDS.toNanos(60)) { + Thread.sleep(50); + } + long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + pool.shutdownNow(); + pool.awaitTermination(15, TimeUnit.SECONDS); + + t.line("== %s", switch (variant) { + case "classic" -> "consumer group (6 KafkaConsumers, same group.id)"; + case "share-default" -> "share group (6 ShareConsumers, default share.acquire.mode)"; + default -> "share group (6 ShareConsumers, share.acquire.mode=record_limit)"; + }); + counts.forEach((id, c) -> t.line("consumer-%d processed %d records", id, c.get())); + t.line("total processed = %d in %d ms", total.get(), millis); + return new Result(counts, millis); + } + + private void consume(String variant, String topic, String group, int id, ConcurrentHashMap counts, + AtomicInteger total) { + Properties p = new Properties(); + p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Broker.BOOTSTRAP); + p.put(ConsumerConfig.GROUP_ID_CONFIG, group); + p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "3"); + boolean share = variant.startsWith("share"); + if (variant.equals("share-record-limit")) { + p.put("share.acquire.mode", "record_limit"); + } + try { + if (share) { + p.put("share.acknowledgement.mode", "explicit"); + try (KafkaShareConsumer c = new KafkaShareConsumer<>(p)) { + c.subscribe(List.of(topic)); + while (!Thread.currentThread().isInterrupted()) { + for (ConsumerRecord r : c.poll(Duration.ofMillis(300))) { + Thread.sleep(50); + c.acknowledge(r, AcknowledgeType.ACCEPT); + counts.get(id).incrementAndGet(); + total.incrementAndGet(); + } + c.commitSync(); + } + } + } + else { + p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + try (KafkaConsumer c = new KafkaConsumer<>(p)) { + c.subscribe(List.of(topic)); + while (!Thread.currentThread().isInterrupted()) { + for (ConsumerRecord r : c.poll(Duration.ofMillis(300))) { + Thread.sleep(50); + counts.get(id).incrementAndGet(); + total.incrementAndGet(); + } + c.commitSync(); + } + } + } + } + catch (InterruptedException | org.apache.kafka.common.errors.InterruptException e) { + Thread.currentThread().interrupt(); + } + } +} diff --git a/share-groups/src/test/resources/logback-test.xml b/share-groups/src/test/resources/logback-test.xml new file mode 100644 index 0000000..fab7eaa --- /dev/null +++ b/share-groups/src/test/resources/logback-test.xml @@ -0,0 +1,8 @@ + + + %d{HH:mm:ss.SSS} %-5level %logger{20} - %msg%n + + + + +