This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new d766e485518 KAFKA-20686: Expand integration test suite for share group
DLQ. [2/N] (#22671)
d766e485518 is described below
commit d766e4855184bfd72f0d894e8c6307564c24af30
Author: Sushant Mahajan <[email protected]>
AuthorDate: Fri Jun 26 21:40:09 2026 +0530
KAFKA-20686: Expand integration test suite for share group DLQ. [2/N]
(#22671)
In this PR we have added integ tests for share group DLQ to include more
scenarios:
* gating DLQ (enable/disable)
* copy record enabled when DLQ
* mixed ack type responses from consumer (covers coalescing)
* dynamic toggling of DLQ
* multi partition DLQ (source and DLQ topic).
Reviewers: Andrew Schofield <[email protected]>
---
.../clients/consumer/ShareConsumerDLQTest.java | 368 +++++++++++++++++++--
1 file changed, 340 insertions(+), 28 deletions(-)
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
index e67d66b0d73..9abb85fc1b9 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
@@ -20,6 +20,8 @@ import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;
@@ -104,10 +106,10 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
produceMessages(recordCount);
// Reject every record using an EXPLICIT-mode share consumer.
- rejectAllRecords(groupId, recordCount);
+ rejectRecords(groupId, recordCount);
// Verify by reading from the DLQ topic, then assert the DLQ metrics.
- verifyDlqTopicRecords(dlqTopic, groupId,
expectedSourceOffsets(recordCount));
+ verifyDlqTopicRecords(dlqTopic, groupId,
expectedSourceOffsets(recordCount), false);
verifyDlqMetrics(groupId, recordCount);
}
@@ -133,11 +135,11 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
produceMessages(recordCount);
- rejectAllRecords(groupId, recordCount);
+ rejectRecords(groupId, recordCount);
// Verify the DLQ topic was auto-created (with DLQ enabled), received
the records, and metrics fired.
verifyDlqTopicCreated(dlqTopic);
- verifyDlqTopicRecords(dlqTopic, groupId,
expectedSourceOffsets(recordCount));
+ verifyDlqTopicRecords(dlqTopic, groupId,
expectedSourceOffsets(recordCount), false);
verifyDlqMetrics(groupId, recordCount);
}
@@ -166,29 +168,278 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
// Repeatedly release the records until their delivery count is
exceeded and they are written to the DLQ.
releaseRecordsUntilDlq(groupId, recordCount);
- verifyDlqTopicRecords(dlqTopic, groupId,
expectedSourceOffsets(recordCount));
+ verifyDlqTopicRecords(dlqTopic, groupId,
expectedSourceOffsets(recordCount), false);
verifyDlqMetrics(groupId, recordCount);
}
- // Consumes from the source topic in EXPLICIT acknowledgement mode and
rejects every record.
- private void rejectAllRecords(String groupId, int recordCount) {
+ /**
+ * Produces records and applies a mix of acknowledgement types so the DLQ
is reached via more than one path:
+ * some records are rejected (DLQ via client reject), some are released on
every delivery so they reach the
+ * DLQ once their delivery count is exceeded (release-based DLQ), and the
rest are accepted (never DLQ'd).
+ * The actions are interleaved by offset so the DLQ'd offsets are
non-contiguous and produce multiple
+ * separate DLQ writes to the same DLQ partition. The DLQ topic is created
manually up front.
+ */
+ @ClusterTest
+ public void testMixedAcknowledgementTypesWrittenToDlq() throws Exception {
+ String groupId = "dlq-mixed-group";
+ // The broker's default share-group DLQ topic prefix is "dlq.", so the
topic name must start with it.
+ String dlqTopic = "dlq.mixed";
+ int recordCount = 6;
+
+ alterShareAutoOffsetReset(groupId, "earliest");
+ // Low delivery count limit so the released records reach the DLQ
(delivery count exceeded) quickly.
+ alterShareDeliveryCountLimit(groupId, "2");
+ createDlqTopic(dlqTopic);
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+
+ produceMessages(recordCount);
+
+ // Assign each source offset an action, interleaved so the DLQ'd
offsets are non-contiguous:
+ // offset % 3 == 0 -> reject (DLQ via client reject)
+ // offset % 3 == 1 -> release (DLQ via delivery count exceeded -
released on every delivery)
+ // offset % 3 == 2 -> accept (never DLQ'd)
+ Set<Long> rejectOffsets = new HashSet<>();
+ Set<Long> releaseOffsets = new HashSet<>();
+ for (long offset = 0; offset < recordCount; offset++) {
+ if (offset % 3 == 0) {
+ rejectOffsets.add(offset);
+ } else if (offset % 3 == 1) {
+ releaseOffsets.add(offset);
+ }
+ }
+ Set<Long> expectedDlqOffsets = new HashSet<>();
+ expectedDlqOffsets.addAll(rejectOffsets);
+ expectedDlqOffsets.addAll(releaseOffsets);
+
try (ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
groupId, Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
EXPLICIT))) {
shareConsumer.subscribe(Set.of(tp.topic()));
- int rejected = 0;
+ // Keep polling/acknowledging until every rejected and released
record has reached the DLQ. Released
+ // records are released on every delivery, so they are redelivered
until their delivery count is
+ // exceeded and they are archived to the DLQ.
long deadlineMs = System.currentTimeMillis() + DEFAULT_MAX_WAIT_MS;
- while (rejected < recordCount && System.currentTimeMillis() <
deadlineMs) {
+ while (dlqMeterCount(METRIC_DLQ_RECORD_COUNT, groupId) <
expectedDlqOffsets.size()
+ && System.currentTimeMillis() < deadlineMs) {
ConsumerRecords<byte[], byte[]> records =
shareConsumer.poll(Duration.ofMillis(2000));
for (ConsumerRecord<byte[], byte[]> record : records) {
- shareConsumer.acknowledge(record, AcknowledgeType.REJECT);
- rejected++;
+ long offset = record.offset();
+ if (rejectOffsets.contains(offset)) {
+ shareConsumer.acknowledge(record,
AcknowledgeType.REJECT);
+ } else if (releaseOffsets.contains(offset)) {
+ shareConsumer.acknowledge(record,
AcknowledgeType.RELEASE);
+ } else {
+ shareConsumer.acknowledge(record,
AcknowledgeType.ACCEPT);
+ }
}
if (records.count() > 0) {
shareConsumer.commitSync(Duration.ofMillis(10000));
}
}
- assertEquals(recordCount, rejected, "Expected to reject all
produced records");
}
+
+ // Exactly the rejected and released records should be on the DLQ;
accepted records must not be.
+ verifyDlqTopicRecords(dlqTopic, groupId, expectedDlqOffsets, false);
+ verifyDlqMetrics(groupId, expectedDlqOffsets.size());
+ }
+
+ /**
+ * As {@link #testRejectedRecordsWrittenToDlqWithCopyRecordDisabled()},
but with record copy enabled, so
+ * each DLQ record must carry the original key and value (in addition to
the context headers).
+ */
+ @ClusterTest
+ public void testRejectedRecordsWrittenToDlqWithCopyRecordEnabled() throws
Exception {
+ String groupId = "dlq-copy-group";
+ String dlqTopic = "dlq.copy";
+ int recordCount = 5;
+
+ alterShareAutoOffsetReset(groupId, "earliest");
+ createDlqTopic(dlqTopic);
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+ // Enable record copy so the original key/value are written onto the
DLQ record.
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_COPY_RECORD_ENABLE_CONFIG, "true");
+
+ produceMessages(recordCount);
+ rejectRecords(groupId, recordCount);
+
+ verifyDlqTopicRecords(dlqTopic, groupId,
expectedSourceOffsets(recordCount), true);
+ verifyDlqMetrics(groupId, recordCount);
+ }
+
+ /**
+ * Rejects records from a multi-partition source topic and verifies they
are routed to the correct DLQ
+ * partition. The destination partition is {@code sourcePartition %
numDlqPartitions}; with a DLQ topic that
+ * has as many partitions as the source, each source partition maps to the
DLQ partition of the same index.
+ */
+ @ClusterTest
+ public void testDlqRecordsRoutedToCorrectDlqPartition() throws Exception {
+ String groupId = "dlq-routing-group";
+ String sourceTopic = "dlq-routing-source";
+ String dlqTopic = "dlq.routing";
+ int partitions = 2;
+ int recordsPerPartition = 3;
+
+ createTopic(sourceTopic, partitions, 1);
+ alterShareAutoOffsetReset(groupId, "earliest");
+ createDlqTopic(dlqTopic, partitions);
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+
+ // Produce the same number of records to each source partition.
+ for (int sourcePartition = 0; sourcePartition < partitions;
sourcePartition++) {
+ produceTo(sourceTopic, sourcePartition, recordsPerPartition);
+ }
+
+ int total = partitions * recordsPerPartition;
+ rejectRecords(groupId, sourceTopic, total);
+ verifyDlqMetrics(groupId, total);
+
+ // Source partition p maps to DLQ partition (p % partitions) == p.
Each DLQ partition should hold the
+ // records from the matching source partition (offsets
0..recordsPerPartition-1).
+ for (int sourcePartition = 0; sourcePartition < partitions;
sourcePartition++) {
+ List<ConsumerRecord<byte[], byte[]>> dlqRecords =
readDlqPartition(dlqTopic, sourcePartition, recordsPerPartition);
+ assertEquals(recordsPerPartition, dlqRecords.size(),
+ "DLQ partition " + sourcePartition + " has an unexpected
number of records");
+ Set<Long> offsets = new HashSet<>();
+ for (ConsumerRecord<byte[], byte[]> record : dlqRecords) {
+ assertEquals(groupId, headerValue(record,
HEADER_DLQ_ERRORS_GROUP));
+ assertEquals(sourceTopic, headerValue(record,
HEADER_DLQ_ERRORS_TOPIC));
+ assertEquals(Integer.toString(sourcePartition),
headerValue(record, HEADER_DLQ_ERRORS_PARTITION),
+ "Records on DLQ partition " + sourcePartition + " must
originate from source partition " + sourcePartition);
+
offsets.add(Long.parseLong(Objects.requireNonNull(headerValue(record,
HEADER_DLQ_ERRORS_OFFSET))));
+ }
+ assertEquals(expectedSourceOffsets(recordsPerPartition), offsets);
+ }
+ }
+
+ /**
+ * Rejects a larger batch of records and verifies they all reach the DLQ.
Exercises the produce coalescing /
+ * record-merge path at volume (many records DLQ'd to the same DLQ
partition).
+ */
+ @ClusterTest
+ public void testManyRejectedRecordsAllWrittenToDlq() throws Exception {
+ String groupId = "dlq-scale-group";
+ String dlqTopic = "dlq.scale";
+ int recordCount = 500;
+
+ alterShareAutoOffsetReset(groupId, "earliest");
+ createDlqTopic(dlqTopic);
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+
+ produceMessages(recordCount);
+ rejectRecords(groupId, recordCount);
+
+ verifyDlqTopicRecords(dlqTopic, groupId,
expectedSourceOffsets(recordCount), false);
+ verifyDlqMetrics(groupId, recordCount);
+ }
+
+ /**
+ * Verifies the DLQ is NOT written when it is gated off, across the gating
conditions:
+ * (a) no DLQ topic configured for the group;
+ * (b) the configured DLQ topic exists but is not DLQ-enabled;
+ * (c) the configured DLQ topic does not exist and auto creation is
disabled (the cluster default);
+ * (d) the configured DLQ topic name does not match the broker's DLQ
topic name prefix ("dlq.").
+ * In every case the rejected records are archived (the start offset
advances) but no DLQ record is written.
+ */
+ @ClusterTest
+ public void testDlqNotTriggeredWhenGatedOff() throws Exception {
+ int recordCount = 3;
+
+ // (a) No DLQ topic configured for the group.
+ assertNoDlqWritten("dlq-gate-noname", "dlq-gate-source-a",
recordCount, group -> { });
+
+ // (b) DLQ topic exists but is not DLQ-enabled.
+ assertNoDlqWritten("dlq-gate-disabled", "dlq-gate-source-b",
recordCount, group -> {
+ createTopic("dlq.disabled");
+ alterShareGroupConfig(group,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, "dlq.disabled");
+ });
+
+ // (c) DLQ topic missing and auto creation disabled (broker default).
+ assertNoDlqWritten("dlq-gate-missing", "dlq-gate-source-c",
recordCount, group ->
+ alterShareGroupConfig(group,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, "dlq.missing"));
+
+ // (d) DLQ topic is DLQ-enabled but its name violates the "dlq."
prefix.
+ assertNoDlqWritten("dlq-gate-prefix", "dlq-gate-source-d",
recordCount, group -> {
+ createDlqTopic("wrong.prefix");
+ alterShareGroupConfig(group,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, "wrong.prefix");
+ });
+ }
+
+ /**
+ * A single consumer rejects records while the DLQ is enabled (they are
written to the DLQ), then the DLQ is
+ * turned off for the group by clearing its DLQ topic name, and the same
consumer keeps rejecting more
+ * records. Verifies via the DLQ record-count metric that no further DLQ
records are written after the DLQ
+ * is disabled.
+ */
+ @ClusterTest
+ public void testDlqStopsAfterDisablingForGroup() throws Exception {
+ String groupId = "dlq-toggle-group";
+ String dlqTopic = "dlq.toggle";
+ int firstBatch = 5;
+ int secondBatch = 5;
+
+ alterShareAutoOffsetReset(groupId, "earliest");
+ createDlqTopic(dlqTopic);
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+
+ try (ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ groupId, Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
EXPLICIT))) {
+ shareConsumer.subscribe(Set.of(tp.topic()));
+
+ // Phase 1: DLQ enabled - reject the first batch; they are written
to the DLQ.
+ produceMessages(firstBatch);
+ rejectRecords(shareConsumer, firstBatch);
+ waitForCondition(() -> dlqMeterCount(METRIC_DLQ_RECORD_COUNT,
groupId) == firstBatch,
+ DEFAULT_MAX_WAIT_MS, 200L,
+ () -> "First batch not written to DLQ, count was " +
dlqMeterCount(METRIC_DLQ_RECORD_COUNT, groupId));
+
+ // Turn DLQ off for the group by clearing its DLQ topic name.
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, "");
+
+ // Phase 2: DLQ disabled - the same consumer rejects more records;
none should be DLQ'd.
+ produceMessages(secondBatch);
+ rejectRecords(shareConsumer, secondBatch);
+
+ // The second-batch records are archived (terminal) and not
redelivered; confirming this also gives
+ // any (erroneous) DLQ write a chance to land before we assert.
+ for (int i = 0; i < 3; i++) {
+ assertEquals(0,
shareConsumer.poll(Duration.ofMillis(2000)).count(),
+ "Archived records must not be redelivered");
+ }
+ }
+
+ // Only the first batch (written while DLQ was enabled) should be on
the DLQ.
+ assertEquals(firstBatch, dlqMeterCount(METRIC_DLQ_RECORD_COUNT,
groupId),
+ "No DLQ records should be written after the DLQ is disabled for
the group");
+ }
+
+ // Consumes from the base source topic in EXPLICIT acknowledgement mode
and rejects `recordCount` records.
+ private void rejectRecords(String groupId, int recordCount) {
+ rejectRecords(groupId, tp.topic(), recordCount);
+ }
+
+ // Consumes from the given source topic in EXPLICIT acknowledgement mode
and rejects `recordCount` records.
+ private void rejectRecords(String groupId, String topic, int recordCount) {
+ try (ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ groupId, Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
EXPLICIT))) {
+ shareConsumer.subscribe(Set.of(topic));
+ rejectRecords(shareConsumer, recordCount);
+ }
+ }
+
+ // Polls the given (already subscribed) consumer and rejects records until
`count` have been rejected.
+ private void rejectRecords(ShareConsumer<byte[], byte[]> shareConsumer,
int count) {
+ int rejected = 0;
+ long deadlineMs = System.currentTimeMillis() + DEFAULT_MAX_WAIT_MS;
+ while (rejected < count && System.currentTimeMillis() < deadlineMs) {
+ ConsumerRecords<byte[], byte[]> records =
shareConsumer.poll(Duration.ofMillis(2000));
+ for (ConsumerRecord<byte[], byte[]> record : records) {
+ shareConsumer.acknowledge(record, AcknowledgeType.REJECT);
+ rejected++;
+ }
+ if (records.count() > 0) {
+ shareConsumer.commitSync(Duration.ofMillis(10000));
+ }
+ }
+ assertEquals(count, rejected, "Expected to reject the requested number
of records");
}
// Repeatedly polls and releases every record until the delivery count
limit is exceeded for all of them
@@ -211,6 +462,34 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
}
}
+ // Sets up a gating condition for a fresh group/source topic, rejects
records, and asserts that although the
+ // records are archived (start offset advances) no DLQ record is ever
written for the group. The setup
+ // callback receives the group id (java.util.function.Consumer is fully
qualified to avoid clashing with the
+ // Kafka Consumer type in this package).
+ private void assertNoDlqWritten(String groupId, String sourceTopic, int
recordCount,
+ java.util.function.Consumer<String>
gateSetup) throws Exception {
+ createTopic(sourceTopic);
+ alterShareAutoOffsetReset(groupId, "earliest");
+ gateSetup.accept(groupId);
+
+ produceTo(sourceTopic, 0, recordCount);
+ rejectRecords(groupId, sourceTopic, recordCount);
+
+ // The rejected records are archived (terminal) so they are never
redelivered. Confirming this also
+ // gives any DLQ attempt time to run (and, for the gated cases, to
fail/short-circuit).
+ try (ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ groupId, Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
EXPLICIT))) {
+ shareConsumer.subscribe(Set.of(sourceTopic));
+ for (int i = 0; i < 3; i++) {
+ assertEquals(0,
shareConsumer.poll(Duration.ofMillis(2000)).count(),
+ "Archived records must not be redelivered for group: " +
groupId);
+ }
+ }
+ // No DLQ record was written for this group (the per-group meter is
never even registered).
+ assertEquals(-1L, dlqMeterCount(METRIC_DLQ_RECORD_COUNT, groupId),
+ "DLQ should not have been written for a gated-off group: " +
groupId);
+ }
+
private static Set<Long> expectedSourceOffsets(int recordCount) {
Set<Long> offsets = new HashSet<>();
for (long offset = 0; offset < recordCount; offset++) {
@@ -245,30 +524,32 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
}
/**
- * Reads the DLQ topic and asserts it received exactly one record per
expected source offset. Record copy
- * is disabled, so each DLQ record carries only the context headers (no
key/value).
+ * Reads the DLQ topic (single partition) and asserts it received exactly
one record per expected source
+ * offset, each carrying the context headers for the base source
topic-partition. When {@code copyEnabled}
+ * is true the DLQ records must also carry the original key/value;
otherwise they carry headers only.
*
* @param dlqTopic the DLQ topic to read from (single
partition)
* @param groupId the share group the rejected records
belonged to
* @param expectedSourceOffsets the source offsets expected to have been
written to the DLQ
+ * @param copyEnabled whether record copy is enabled for the
group
*/
- private void verifyDlqTopicRecords(String dlqTopic, String groupId,
Set<Long> expectedSourceOffsets) throws InterruptedException {
- TopicPartition dlqTp = new TopicPartition(dlqTopic, 0);
- List<ConsumerRecord<byte[], byte[]>> dlqRecords = new ArrayList<>();
- try (Consumer<byte[], byte[]> consumer = cluster.consumer()) {
- consumer.assign(List.of(dlqTp));
- consumer.seekToBeginning(List.of(dlqTp));
- waitForCondition(() -> {
-
dlqRecords.addAll(consumer.poll(Duration.ofMillis(1000)).records(dlqTp));
- return dlqRecords.size() >= expectedSourceOffsets.size();
- }, DEFAULT_MAX_WAIT_MS, 500L, () -> "DLQ topic did not receive " +
expectedSourceOffsets.size() + " records, got " + dlqRecords.size());
- }
+ private void verifyDlqTopicRecords(String dlqTopic, String groupId,
Set<Long> expectedSourceOffsets,
+ boolean copyEnabled) throws
InterruptedException {
+ List<ConsumerRecord<byte[], byte[]>> dlqRecords =
readDlqPartition(dlqTopic, 0, expectedSourceOffsets.size());
assertEquals(expectedSourceOffsets.size(), dlqRecords.size(),
"Unexpected number of records on the DLQ topic");
Set<Long> actualSourceOffsets = new HashSet<>();
for (ConsumerRecord<byte[], byte[]> record : dlqRecords) {
- // Record copy is disabled, so only headers are written - the
value (and key) are null.
- assertNull(record.value(), "DLQ record value should be null when
record copy is disabled");
+ if (copyEnabled) {
+ // produceMessages() produces records with key "key" and value
"value".
+ assertEquals("key", new
String(Objects.requireNonNull(record.key()), StandardCharsets.UTF_8),
+ "DLQ record key should be copied when record copy is
enabled");
+ assertEquals("value", new
String(Objects.requireNonNull(record.value()), StandardCharsets.UTF_8),
+ "DLQ record value should be copied when record copy is
enabled");
+ } else {
+ // Record copy is disabled, so only headers are written - the
value (and key) are null.
+ assertNull(record.value(), "DLQ record value should be null
when record copy is disabled");
+ }
assertEquals(groupId, headerValue(record,
HEADER_DLQ_ERRORS_GROUP));
assertEquals(tp.topic(), headerValue(record,
HEADER_DLQ_ERRORS_TOPIC));
assertEquals(Integer.toString(tp.partition()), headerValue(record,
HEADER_DLQ_ERRORS_PARTITION));
@@ -277,16 +558,47 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
assertEquals(expectedSourceOffsets, actualSourceOffsets, "DLQ records
should cover every expected source offset");
}
+ // Reads at least `expectedCount` records from a single DLQ
topic-partition.
+ private List<ConsumerRecord<byte[], byte[]>> readDlqPartition(String
dlqTopic, int partition, int expectedCount)
+ throws InterruptedException {
+ TopicPartition dlqTp = new TopicPartition(dlqTopic, partition);
+ List<ConsumerRecord<byte[], byte[]>> dlqRecords = new ArrayList<>();
+ try (Consumer<byte[], byte[]> consumer = cluster.consumer()) {
+ consumer.assign(List.of(dlqTp));
+ consumer.seekToBeginning(List.of(dlqTp));
+ waitForCondition(() -> {
+
dlqRecords.addAll(consumer.poll(Duration.ofMillis(1000)).records(dlqTp));
+ return dlqRecords.size() >= expectedCount;
+ }, DEFAULT_MAX_WAIT_MS, 500L,
+ () -> dlqTp + " did not receive " + expectedCount + " records,
got " + dlqRecords.size());
+ }
+ return dlqRecords;
+ }
+
private void createDlqTopic(String topicName) {
+ createDlqTopic(topicName, 1);
+ }
+
+ private void createDlqTopic(String topicName, int numPartitions) {
assertDoesNotThrow(() -> {
try (Admin admin = createAdminClient()) {
- NewTopic newTopic = new NewTopic(topicName, 1, (short) 1)
+ NewTopic newTopic = new NewTopic(topicName, numPartitions,
(short) 1)
.configs(Map.of(TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG,
"true"));
admin.createTopics(Set.of(newTopic)).all().get();
}
}, "Failed to create DLQ topic");
}
+ // Produces `count` records (key "key", value "value") to a specific
topic-partition.
+ private void produceTo(String topic, int partition, int count) {
+ try (Producer<byte[], byte[]> producer = createProducer()) {
+ for (int i = 0; i < count; i++) {
+ producer.send(new ProducerRecord<>(topic, partition,
"key".getBytes(), "value".getBytes()));
+ }
+ producer.flush();
+ }
+ }
+
private static String headerValue(ConsumerRecord<byte[], byte[]> record,
String key) {
Header header = record.headers().lastHeader(key);
return header == null ? null : new String(header.value(),
StandardCharsets.UTF_8);
