Croway commented on PR #28730:
URL: https://github.com/apache/camel/pull/28730#issuecomment-6544237001
@ramu11 here it is. Drop it into
`components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/`
and run:
```
mvn -pl components/camel-kafka-common,components/camel-kafka verify
-Dit.test=KafkaExactlyOnceReReviewIT -Dtest=skip
-Dsurefire.failIfNoSpecifiedTests=false
```
It uses the test-infra Kafka container. Each test creates its own
CamelContext, topics, group and `transactionalId`, and asserts the **correct**
exactly-once outcome, so a failing test is a remaining problem. Lines starting
with `REPRO` in `target/camel-kafka-test.log` hold what each test observed
(attempts per record key, committed output, committed source offset, failures).
| Test | Scenario | At `f9d92812` |
|---|---|---|
| `o1`..`o9` | first review round (rollbackOnly, removeHeaders, split +
aggregation, seda after the producer, DLC without redelivery, filter, producer
without exactlyOnce, startup rejections, wireTap next to a failing send) | pass
|
| `n1`, `n2` | Kafka DLQ topic, plain or exactly-once endpoint (point 1) |
**fail**: offset committed, DLQ topic empty |
| `n3` | DLC with `maximumRedeliveries(2)` | pass (record skipped, DLQ once)
|
| `n4`, `n5` | seda or wireTap as the only path to Kafka (point 2) |
**fail**: offset committed, output empty |
| `n6` | split fan-out, each child sends (point 3) | **fail**: replays
forever |
| `n7` | two routes sharing one sink URI / `transactionalId` | pass, logs
the fencing failure |
| `n8` | split + `groupedBody()`, then one send (the fan-out pattern that
works) | pass |
`n1`, `n2`, `n4`, `n5` and `n6` should pass once the three points are
addressed. Feel free to take whatever is useful into the PR's own ITs.
<details>
<summary><code>KafkaExactlyOnceReReviewIT.java</code></summary>
```java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.camel.component.kafka.integration;
import java.time.Duration;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.AggregationStrategies;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaComponent;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.spi.CamelEvent;
import org.apache.camel.support.EventNotifierSupport;
import org.apache.camel.test.infra.kafka.services.KafkaService;
import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
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.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.awaitility.core.ConditionTimeoutException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
/**
* Exactly-once scenarios from the review of PR #28730. Every test asserts
the correct exactly-once outcome, so a
* failing test is a remaining problem. Lines starting with "REPRO" in
target/camel-kafka-test.log hold the observed
* values (attempts per record key, committed output, committed source
offset, failures).
* <p>
* o1..o9 cover the first review round and pass at f9d92812. n1..n8 are the
second round: at f9d92812 n1, n2 (Kafka DLQ
* topic), n4, n5 (seda/wireTap as the only path to Kafka) and n6 (split
fan-out) fail; n3, n7 and n8 pass, and n7 logs
* the fencing between two producers sharing one transactionalId.
* <p>
* Run with: mvn -pl components/camel-kafka-common,components/camel-kafka
verify -Dit.test=KafkaExactlyOnceReReviewIT
* -Dtest=skip -Dsurefire.failIfNoSpecifiedTests=false
*/
class KafkaExactlyOnceReReviewIT {
private static final Logger LOG =
LoggerFactory.getLogger(KafkaExactlyOnceReReviewIT.class);
@RegisterExtension
static KafkaService service =
KafkaServiceFactory.createSingletonService();
private final Map<String, AtomicInteger> attempts = new
ConcurrentHashMap<>();
private final List<String> failures = new CopyOnWriteArrayList<>();
private final List<CamelContext> contexts = new ArrayList<>();
private AdminClient admin;
private String id;
@BeforeEach
void setUp(TestInfo info) {
id = info.getTestMethod().orElseThrow().getName() + "-" +
Long.toHexString(System.nanoTime());
admin =
AdminClient.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, boot()));
}
@AfterEach
void tearDown() {
for (CamelContext context : contexts) {
context.stop();
}
admin.close();
}
//
=================================================================================================================
// Original findings: expected to be fixed
//
=================================================================================================================
@Test
void o1MarkRollbackOnlyReplaysTheRecord() throws Exception {
String in = topics("in", "out")[0];
AtomicBoolean rolledBack = new AtomicBoolean();
start(new RouteBuilder() {
@Override
public void configure() {
from(source(in)).routeId(id).process(countAttempt()).to(sink(id + "-out", "tx"))
.filter(e -> "m1".equals(key(e)) &&
rolledBack.compareAndSet(false, true))
.markRollbackOnly().end();
}
});
produce(in, "m0", "m1", "m2");
awaitCommitted(in, 3);
List<String> out = readCommittedKeys(id + "-out");
LOG.info("REPRO o1 rollbackOnly: attempts={} out={} committed={}",
attempts, out, committed(id, in));
assertEquals(List.of("m0", "m1", "m2"), out);
assertEquals(2, attempts("m1"));
}
@Test
void o2RemoveHeadersKeepsTheOffsets() throws Exception {
String in = topics("in", "out")[0];
start(new RouteBuilder() {
@Override
public void configure() {
from(source(in)).routeId(id).process(countAttempt()).removeHeaders("*").to(sink(id
+ "-out", "tx"));
}
});
produce(in, "m0", "m1");
awaitCommitted(in, 2);
List<String> out = readCommittedValues(id + "-out");
LOG.info("REPRO o2 removeHeaders: attempts={} out={}", attempts,
out);
assertEquals(List.of("m0", "m1"), out);
}
@Test
void o3SplitWithAggregationStrategySendsTheParent() throws Exception {
String in = topics("in", "out")[0];
start(new RouteBuilder() {
@Override
public void configure() {
from(source(in)).routeId(id).process(countAttempt())
.split(body().tokenize(","),
AggregationStrategies.useLatest()).transform(body().append("!")).end()
.to(sink(id + "-out", "tx"));
}
});
produce(in, Map.of("m0", "a,b,c"));
awaitCommitted(in, 1);
List<String> out = readCommittedValues(id + "-out");
LOG.info("REPRO o3 split+aggregation: attempts={} out={}", attempts,
out);
assertEquals(List.of("c!"), out);
assertEquals(1, attempts("m0"));
}
@Test
void o4SedaAfterTheProducerDoesNotWedgeTheRoute() throws Exception {
String in = topics("in", "out")[0];
start(new RouteBuilder() {
@Override
public void configure() {
from(source(in)).routeId(id).process(countAttempt()).to(sink(id + "-out",
"tx")).to("seda:audit");
from("seda:audit").to("log:audit?level=DEBUG");
}
});
produce(in, "m0", "m1", "m2");
awaitCommitted(in, 3);
List<String> out = readCommittedKeys(id + "-out");
LOG.info("REPRO o4 seda after producer: attempts={} out={}
failures={}", attempts, out, sample());
assertEquals(List.of("m0", "m1", "m2"), out);
}
@Test
void o5DeadLetterChannelWithoutRedeliverySkipsThePoisonRecord() throws
Exception {
String in = topics("in", "out")[0];
AtomicInteger dlq = new AtomicInteger();
start(new RouteBuilder() {
@Override
public void configure() {
errorHandler(deadLetterChannel("direct:dlq"));
from("direct:dlq").process(e -> dlq.incrementAndGet());
from(source(in)).routeId(id).process(countAttempt())
.to(sink(id + "-out", "tx") +
"&maxRequestSize=1024");
}
});
produce(in, poisonRecords());
awaitCommitted(in, 3);
List<String> out = readCommittedKeys(id + "-out");
LOG.info("REPRO o5 DLC no redelivery: attempts={} dlq={} out={}",
attempts, dlq.get(), out);
assertEquals(List.of("small-0", "small-2"), out);
assertEquals(1, dlq.get());
}
@Test
void o6FilteredRecordsCommitTheirOffsets() throws Exception {
String in = topics("in", "out")[0];
CamelContext context = start(new RouteBuilder() {
@Override
public void configure() {
from(source(in)).routeId(id).process(countAttempt())
.filter(header(KafkaConstants.KEY).startsWith("m")).to(sink(id + "-out", "tx"));
}
});
produce(in, "m0", "skip1", "skip2", "skip3");
awaitCommitted(in, 4);
context.getRouteController().stopRoute(id);
context.getRouteController().startRoute(id);
produce(in, "m4");
awaitCommitted(in, 5);
LOG.info("REPRO o6 filter: attempts={}", attempts);
assertEquals(1, attempts("skip1"), "not consumed again after the
restart");
}
@Test
void o7ProducerWithoutExactlyOnceIsRejectedAndTheRecordReplays() throws
Exception {
String in = topics("in", "out")[0];
start(new RouteBuilder() {
@Override
public void configure() {
from(source(in)).routeId(id).process(countAttempt()).to("kafka:" + id +
"-out?transacted=true");
}
});
produce(in, "m0");
await().atMost(60, TimeUnit.SECONDS).until(() -> attempts("m0") >=
2);
LOG.info("REPRO o7 producer without exactlyOnce: attempts={}
committed={} out={} failures={}", attempts,
committed(id, in), readCommittedKeys(id + "-out"), sample());
assertEquals(null, committed(id, in));
assertTrue(readCommittedKeys(id + "-out").isEmpty());
}
@Test
void o8RepositioningOptionsAreRejectedAtStartup() throws Exception {
String in = topics("in")[0];
for (String extra : List.of("&seekTo=BEGINNING", "&seekTo=END",
"&batching=true", "&consumersCount=2")) {
CamelContext context = newContext();
context.addRoutes(new RouteBuilder() {
@Override
public void configure() {
from(source(in) + extra).to(sink(id + "-out", "tx"));
}
});
Exception e = assertThrows(Exception.class, context::start,
extra);
LOG.info("REPRO o8 {} rejected: {}", extra, rootMessage(e));
context.stop();
}
}
@Test
void o9WireTapNextToTheMainSendCannotSkipAFailedRecord() throws
Exception {
String in = topics("in", "out", "audit")[0];
AtomicBoolean failed = new AtomicBoolean();
start(new RouteBuilder() {
@Override
public void configure() {
from(source(in)).routeId(id).process(countAttempt())
.wireTap("direct:audit")
.to(sink(id + "-out", "tx"))
.delay(1000)
.process(failOnce("m1", failed));
from("direct:audit").to(sink(id + "-audit", "audit"));
}
});
produce(in, "m0", "m1", "m2");
awaitCommitted(in, 3);
List<String> out = readCommittedKeys(id + "-out");
LOG.info("REPRO o9 wireTap next to main send: attempts={} out={}
audit={} failures={}", attempts, out,
readCommittedKeys(id + "-audit"), sample());
assertEquals(List.of("m0", "m1", "m2"), out);
}
//
=================================================================================================================
// New probes
//
=================================================================================================================
/** A Kafka DLQ topic is the usual dead letter channel of a Kafka route.
*/
@Test
void n1DeadLetterChannelToAKafkaTopic() throws Exception {
deadLetterToKafka("kafka:" + id + "-dlq");
}
@Test
void n2DeadLetterChannelToAnExactlyOnceKafkaTopic() throws Exception {
deadLetterToKafka("kafka:" + id +
"-dlq?exactlyOnce=true&transactionalId=" + id + "-dlq-tx");
}
private void deadLetterToKafka(String dlqUri) throws Exception {
String in = topics("in", "out", "dlq")[0];
start(new RouteBuilder() {
@Override
public void configure() {
errorHandler(deadLetterChannel(dlqUri));
from(source(in)).routeId(id).process(countAttempt())
.to(sink(id + "-out", "tx") +
"&maxRequestSize=1024");
}
});
produce(in, poisonRecords());
boolean committed3 = awaitQuietly(() ->
Long.valueOf(3).equals(committed(id, in)), 45);
List<String> out = readCommittedKeys(id + "-out");
List<String> dlq = readCommittedKeys(id + "-dlq");
LOG.info("REPRO {}: dlqUri={} committedOffset={} out={} dlqTopic={}
attempts={} failures={}", id, dlqUri,
committed(id, in), out, dlq, attempts, sample());
if (committed3) {
assertTrue(dlq.contains("big-1"), "the offset moved past big-1,
so it must be in the DLQ topic: " + dlq);
} else {
fail("the route never moved past big-1: attempts=" + attempts);
}
}
/** A dead letter channel usually redelivers a few times first. */
@Test
void n3DeadLetterChannelWithRedeliveries() throws Exception {
String in = topics("in", "out")[0];
AtomicInteger dlq = new AtomicInteger();
start(new RouteBuilder() {
@Override
public void configure() {
errorHandler(deadLetterChannel("direct:dlq").maximumRedeliveries(2).redeliveryDelay(0));
from("direct:dlq").process(e -> dlq.incrementAndGet());
from(source(in)).routeId(id).process(countAttempt())
.to(sink(id + "-out", "tx") +
"&maxRequestSize=1024");
}
});
produce(in, poisonRecords());
boolean committed3 = awaitQuietly(() ->
Long.valueOf(3).equals(committed(id, in)), 45);
LOG.info("REPRO n3 DLC maximumRedeliveries(2): committed3={}
committedOffset={} attempts={} dlq={} out={}"
+ " failures={}",
committed3, committed(id, in), attempts, dlq.get(),
readCommittedKeys(id + "-out"), sample());
assertTrue(committed3, "the poison record is never skipped:
attempts=" + attempts + " dlq=" + dlq.get());
assertEquals(1, dlq.get());
}
/** The only Kafka send is in a SEDA route. */
@Test
void n4SedaIsTheOnlyPathToKafka() throws Exception {
onlyKafkaPathIsACopy(true);
}
/** The only Kafka send is behind a wireTap. */
@Test
void n5WireTapIsTheOnlyPathToKafka() throws Exception {
onlyKafkaPathIsACopy(false);
}
private void onlyKafkaPathIsACopy(boolean seda) throws Exception {
String in = topics("in", "out")[0];
start(new RouteBuilder() {
@Override
public void configure() {
if (seda) {
from(source(in)).routeId(id).process(countAttempt()).to("seda:out");
from("seda:out").to(sink(id + "-out", "tx"));
} else {
from(source(in)).routeId(id).process(countAttempt()).wireTap("direct:out").to("log:main?level=DEBUG");
from("direct:out").to(sink(id + "-out", "tx"));
}
}
});
produce(in, "m0", "m1", "m2");
boolean committed3 = awaitQuietly(() ->
Long.valueOf(3).equals(committed(id, in)), 30);
List<String> out = readCommittedKeys(id + "-out");
LOG.info("REPRO {}: committedOffset={} out={} attempts={}
failures={}", id, committed(id, in), out, attempts,
sample());
if (committed3) {
assertEquals(List.of("m0", "m1", "m2"), out, "offsets committed,
so the outputs must exist");
}
}
/** One input record fans out into several output records. */
@Test
void n6SplitFanOutToKafka() throws Exception {
String in = topics("in", "out")[0];
start(new RouteBuilder() {
@Override
public void configure() {
from(source(in)).routeId(id).process(countAttempt())
.split(body().tokenize(",")).to(sink(id + "-out",
"tx")).end();
}
});
produce(in, Map.of("m0", "a,b,c"));
boolean committed1 = awaitQuietly(() ->
Long.valueOf(1).equals(committed(id, in)), 30);
List<String> out = readCommittedValues(id + "-out");
LOG.info("REPRO n6 split fan-out: committed1={} committedOffset={}
out={} attempts={} failures={}", committed1,
committed(id, in), out, attempts, sample());
if (committed1) {
assertEquals(List.of("a", "b", "c"), out, "offset committed, so
every split output must exist once");
}
}
/** Workaround for the fan-out: one send of an iterable body, so every
record is in one transaction. */
@Test
void n8FanOutAsOneIterableSend() throws Exception {
String in = topics("in", "out")[0];
start(new RouteBuilder() {
@Override
public void configure() {
from(source(in)).routeId(id).process(countAttempt())
.split(body().tokenize(","),
AggregationStrategies.groupedBody()).transform(body().append("!")).end()
.to(sink(id + "-out", "tx"));
}
});
produce(in, Map.of("m0", "a,b,c"));
boolean committed1 = awaitQuietly(() ->
Long.valueOf(1).equals(committed(id, in)), 30);
List<String> out = readCommittedValues(id + "-out");
LOG.info("REPRO n8 fan-out as one iterable send: committed1={}
out={} attempts={} failures={}", committed1, out,
attempts, sample());
assertTrue(committed1);
assertEquals(List.of("a!", "b!", "c!"), out);
}
/** Two routes write to the same sink endpoint URI, so to the same
transactionalId. */
@Test
void n7TwoRoutesShareOneSinkUri() throws Exception {
String in1 = topics("in1", "in2", "out")[0];
String in2 = id + "-in2";
String sink = sink(id + "-out", "shared");
start(new RouteBuilder() {
@Override
public void configure() {
from(source(in1, id + "-g1")).routeId(id +
"-r1").process(countAttempt()).to(sink);
from(source(in2, id + "-g2")).routeId(id +
"-r2").process(countAttempt()).to(sink);
}
});
String[] a = new String[10];
String[] b = new String[10];
for (int i = 0; i < 10; i++) {
a[i] = "a" + i;
b[i] = "b" + i;
}
long start = System.currentTimeMillis();
produce(in1, a);
produce(in2, b);
boolean drained = awaitQuietly(() -> readCommittedKeys(id +
"-out").size() >= 20, 90);
List<String> out = readCommittedKeys(id + "-out");
long fenced = failures.stream().filter(f -> f.contains("Fenced") ||
f.contains("fenced")).count();
LOG.info("REPRO n7 shared transactionalId: drained={} in {} ms
out={} (distinct {}) attempts={} fencedFailures={}"
+ " failures={}",
drained, System.currentTimeMillis() - start, out.size(),
out.stream().distinct().count(),
attempts.values().stream().mapToInt(AtomicInteger::get).sum(), fenced,
sample());
assertTrue(drained, "not all records delivered");
assertEquals(20, out.size(), "exactly once");
}
//
=================================================================================================================
// helpers
//
=================================================================================================================
private static String boot() {
return service.getBootstrapServers();
}
private String source(String topic) {
return source(topic, id);
}
private static String source(String topic, String group) {
return "kafka:" + topic + "?groupId=" + group +
"&exactlyOnce=true&autoOffsetReset=earliest"
+
"&autoCommitEnable=false&allowManualCommit=true&breakOnFirstError=true&pollTimeoutMs=1000";
}
private String sink(String topic, String txSuffix) {
return "kafka:" + topic + "?exactlyOnce=true&transactionalId=" + id
+ "-" + txSuffix + "&requestRequiredAcks=-1";
}
private static Map<String, String> poisonRecords() {
Map<String, String> records = new LinkedHashMap<>();
records.put("small-0", "small");
records.put("big-1", "x".repeat(4000));
records.put("small-2", "small");
return records;
}
private CamelContext newContext() {
CamelContext context = new DefaultCamelContext();
KafkaComponent kafka = new KafkaComponent(context);
kafka.getConfiguration().setBrokers(boot());
context.addComponent("kafka", kafka);
context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
@Override
public void notify(CamelEvent event) {
if (event instanceof CamelEvent.ExchangeFailedEvent failed) {
failures.add(String.valueOf(failed.getExchange().getException()));
}
}
});
return context;
}
private CamelContext start(RouteBuilder routes) throws Exception {
CamelContext context = newContext();
context.addRoutes(routes);
context.start();
contexts.add(context);
return context;
}
private static String key(Exchange e) {
return e.getMessage().getHeader(KafkaConstants.KEY, String.class);
}
private Processor countAttempt() {
return e -> attempts.computeIfAbsent(key(e), k -> new
AtomicInteger()).incrementAndGet();
}
private static Processor failOnce(String key, AtomicBoolean failed) {
return e -> {
if (key.equals(key(e)) && failed.compareAndSet(false, true)) {
throw new IllegalStateException("failure injected by the
test after the send");
}
};
}
private int attempts(String key) {
AtomicInteger count = attempts.get(key);
return count == null ? 0 : count.get();
}
private List<String> sample() {
return failures.stream().distinct().limit(3).toList();
}
private static String rootMessage(Throwable t) {
while (t.getCause() != null) {
t = t.getCause();
}
return t.getMessage();
}
private static boolean awaitQuietly(Callable<Boolean> condition, int
seconds) {
try {
await().atMost(seconds, TimeUnit.SECONDS).until(condition);
return true;
} catch (ConditionTimeoutException e) {
return false;
}
}
private void awaitCommitted(String topic, long offset) {
await().atMost(60, TimeUnit.SECONDS).until(() ->
Long.valueOf(offset).equals(committed(id, topic)));
}
/** Creates the topics id-name and returns their names. */
private String[] topics(String... names) throws Exception {
String[] topics = new String[names.length];
try (AdminClient client =
AdminClient.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, boot())))
{
List<NewTopic> newTopics = new ArrayList<>();
for (int i = 0; i < names.length; i++) {
topics[i] = id + "-" + names[i];
newTopics.add(new NewTopic(topics[i], 1, (short) 1));
}
client.createTopics(newTopics).all().get(30, TimeUnit.SECONDS);
}
return topics;
}
private static void produce(String topic, String... keys) {
Map<String, String> records = new LinkedHashMap<>();
for (String key : keys) {
records.put(key, key);
}
produce(topic, records);
}
private static void produce(String topic, Map<String, String> records) {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, boot());
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
try (KafkaProducer<String, String> producer = new
KafkaProducer<>(props)) {
records.forEach((k, v) -> producer.send(new
ProducerRecord<>(topic, k, v)));
producer.flush();
}
}
private static List<String> readCommittedKeys(String topic) {
return readCommitted(topic, true);
}
private static List<String> readCommittedValues(String topic) {
return readCommitted(topic, false);
}
private static List<String> readCommitted(String topic, boolean keys) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, boot());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
List<String> result = new ArrayList<>();
try (KafkaConsumer<String, String> consumer = new
KafkaConsumer<>(props)) {
List<TopicPartition> partitions = consumer.partitionsFor(topic,
Duration.ofSeconds(10)).stream()
.map(p -> new TopicPartition(topic,
p.partition())).toList();
consumer.assign(partitions);
consumer.seekToBeginning(partitions);
Map<TopicPartition, Long> end = consumer.endOffsets(partitions);
long deadline = System.currentTimeMillis() + 15000;
while (System.currentTimeMillis() < deadline
&& partitions.stream().anyMatch(tp ->
consumer.position(tp) < end.get(tp))) {
consumer.poll(Duration.ofMillis(200)).forEach(r ->
result.add(keys ? r.key() : r.value()));
}
}
return result;
}
private Long committed(String group, String topic) throws Exception {
OffsetAndMetadata offset =
admin.listConsumerGroupOffsets(group).partitionsToOffsetAndMetadata()
.get(30, TimeUnit.SECONDS).get(new TopicPartition(topic, 0));
return offset == null ? null : offset.offset();
}
}
```
</details>
_Claude Code on behalf of Croway_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
