You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2026/09/09 17:02:44 UTC
(kafka) branch trunk updated: KAFKA-21057: Extract share-group DLQ record and validation logic into reusable helpers (#23360)
This is an automated email from the ASF dual-hosted git repository.
apoorvmittal10 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 a94c2892556 KAFKA-21057: Extract share-group DLQ record and validation logic into reusable helpers (#23360)
a94c2892556 is described below
commit a94c2892556a57ea7215eb168f664753a7fdcd9a
Author: Abhinav Dixit <ad...@confluent.io>
AuthorDate: Wed Sep 9 22:32:37 2026 +0530
KAFKA-21057: Extract share-group DLQ record and validation logic into reusable helpers (#23360)
This is a behavior-preserving refactor that extracts logic out of
`ShareGroupDLQStateManager` into two new shared classes under
`org.apache.kafka.server.share.dlq`
- `ShareGroupDLQRecordHelper`: builds DLQ record headers, resolves the
source
topic name for a `TopicIdPartition`, computes the destination DLQ
partition, builds `MemoryRecords` for a DLQ batch while respecting
`max.message.bytes`, and optionally fetches source records for
copy-record mode.
- `ShareGroupDLQValidator`: validates `ShareGroupDLQRecordParameter`
fields,
and validates DLQ topic configuration.
Also adds an empty, overridable `waitForDlqTopicConfigRefresh()` hook in
`ShareConsumerDLQTest` so subclasses can wait for DLQ topic config
propagation before producing records, without changing behavior for the
existing test.
### Testing
Pure refactor, no intended behavior change. Existing coverage in
`ShareGroupDLQStateManagerTest` and `ShareConsumerDLQTest` continues to
exercise this logic through the same public entry points (only the
static header-constant imports moved to their new home in
`ShareGroupDLQRecordHelper`). No new tests were needed since no new
behavior was introduced.
Reviewers: Apoorv Mittal <ap...@gmail.com>, Sushant Mahajan
<sm...@confluent.io>
---
.../clients/consumer/ShareConsumerDLQTest.java | 6 +
.../share/dlq/DefaultShareGroupDLQManager.java | 37 +--
.../share/dlq/ShareGroupDLQRecordHelper.java | 267 +++++++++++++++++++++
.../share/dlq/ShareGroupDLQStateManager.java | 161 ++-----------
.../server/share/dlq/ShareGroupDLQValidator.java | 106 ++++++++
.../share/dlq/ShareGroupDLQStateManagerTest.java | 12 +-
6 files changed, 403 insertions(+), 186 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 2d5f415a9e5..6776a3d8373 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
@@ -583,6 +583,10 @@ public class ShareConsumerDLQTest extends ShareConsumerTestBase {
}, DEFAULT_MAX_WAIT_MS, 100L, () -> "Raised max.message.bytes did not propagate on the DLQ topic");
}
+ // Hook for any subclasses whose DLQ manager caches topic config (e.g. max.message.bytes)
+ // and needs time for the raised value to propagate before the next produce batch.
+ waitForDlqTopicConfigRefresh();
+
try (Producer<byte[], byte[]> producer = createProducer()) {
for (int i = 0; i < recordCount; i++) {
producer.send(new ProducerRecord<>(sourceTopic, 0, "key".getBytes(StandardCharsets.UTF_8), payload));
@@ -885,6 +889,8 @@ public class ShareConsumerDLQTest extends ShareConsumerTestBase {
"Expected at least one DLQ produce request to have been enqueued");
}
+ protected void waitForDlqTopicConfigRefresh() throws InterruptedException {}
+
// Verifies the DLQ topic was auto-created by the broker and has the DLQ-enable topic config set.
private void verifyDlqTopicCreated(String dlqTopic) throws Exception {
try (Admin admin = createAdminClient()) {
diff --git a/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java b/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
index 1c78a33ae72..a7adf3e7e52 100644
--- a/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
+++ b/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
@@ -72,7 +72,7 @@ public class DefaultShareGroupDLQManager implements ShareGroupDLQManager {
@Override
public CompletableFuture<Void> enqueue(ShareGroupDLQRecordParameter param) {
try {
- validate(param);
+ ShareGroupDLQValidator.validateParam(param);
return stateManager.dlq(param);
} catch (Exception e) {
log.error("Unable to enqueue DLQ request", e);
@@ -88,39 +88,4 @@ public class DefaultShareGroupDLQManager implements ShareGroupDLQManager {
log.error("Unable to stop DLQ state manager", e);
}
}
-
- private static void validate(ShareGroupDLQRecordParameter param) {
- String prefix = "DLQ records parameters";
- if (param == null) {
- throw new IllegalArgumentException(prefix + " cannot be null.");
- }
-
- if (param.groupId() == null || param.groupId().isEmpty()) {
- throw new IllegalArgumentException(prefix + " group cannot be null or empty.");
- }
-
- if (param.topicIdPartition() == null) {
- throw new IllegalArgumentException(prefix + " topic/partition data cannot be null or empty.");
- }
-
- if (param.topicIdPartition().topicId() == null) {
- throw new IllegalArgumentException(prefix + " topic id data cannot be null or empty.");
- }
-
- if (param.topicIdPartition().partition() < 0) {
- throw new IllegalArgumentException(prefix + " partition cannot be negative.");
- }
-
- if (param.firstOffset() < 0) {
- throw new IllegalArgumentException(prefix + " first offset cannot be negative.");
- }
-
- if (param.lastOffset() < 0) {
- throw new IllegalArgumentException(prefix + " last offset cannot be negative.");
- }
-
- if (param.lastOffset() < param.firstOffset()) {
- throw new IllegalArgumentException(prefix + " last offset cannot be less than first offset.");
- }
- }
}
diff --git a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordHelper.java b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordHelper.java
new file mode 100644
index 00000000000..62f18a60ef1
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordHelper.java
@@ -0,0 +1,267 @@
+/*
+ * 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.kafka.server.share.dlq;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.compress.Compression;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.record.internal.DefaultRecord;
+import org.apache.kafka.common.record.internal.DefaultRecordBatch;
+import org.apache.kafka.common.record.internal.MemoryRecords;
+import org.apache.kafka.common.record.internal.Record;
+import org.apache.kafka.common.record.internal.SimpleRecord;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.share.LogReader;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+
+/**
+ * Helper for DLQ record building and source-record fetching.
+ */
+public final class ShareGroupDLQRecordHelper {
+
+ private ShareGroupDLQRecordHelper() {}
+
+ /**
+ * In most cases we expect the records getting DLQ'ed will be single offsets and
+ * not complete batches. Hence, using a large upper limit while reading from the log
+ * would be fruitless in most cases. Therefore, the value of 1 MB has been chosen
+ * for the DLQ-related log reads.
+ */
+ private static final int DLQ_MAX_FETCH_BYTES = 1024 * 1024;
+
+ protected static final String HEADER_DLQ_ERRORS_TOPIC = "__dlq.errors.topic";
+ protected static final String HEADER_DLQ_ERRORS_PARTITION = "__dlq.errors.partition";
+ protected static final String HEADER_DLQ_ERRORS_OFFSET = "__dlq.errors.offset";
+ protected static final String HEADER_DLQ_ERRORS_GROUP = "__dlq.errors.group";
+ protected static final String HEADER_DLQ_ERRORS_DELIVERY_COUNT = "__dlq.errors.delivery.count";
+ protected static final String HEADER_DLQ_ERRORS_MESSAGE = "__dlq.errors.message";
+
+ /**
+ * Result of building DLQ records for a range of offsets, respecting maxMessageBytes.
+ *
+ * @param records The built MemoryRecords containing DLQ records with headers
+ * @param lastOffsetIncluded The last source offset included in this batch
+ * @param recordCount The number of individual records in the batch
+ */
+ public record BuildResult(MemoryRecords records, long lastOffsetIncluded, int recordCount) {
+ }
+
+ /**
+ * Builds DLQ headers for a single offset.
+ *
+ * @param sourceTopic The resolved source topic name
+ * @param partition The source partition number
+ * @param offset The source offset
+ * @param groupId The share group ID
+ * @param deliveryCount Optional delivery count
+ * @param cause Optional cause/reason for DLQ
+ * @return Array of DLQ headers
+ */
+ private static Header[] headers(
+ String sourceTopic,
+ int partition,
+ long offset,
+ String groupId,
+ Optional<Short> deliveryCount,
+ Optional<Throwable> cause
+ ) {
+ String causeMessage = cause.map(Throwable::getMessage).orElse(null);
+ int size = 4 + (deliveryCount.isPresent() ? 1 : 0) + (causeMessage != null ? 1 : 0);
+
+ Header[] headers = new Header[size];
+ int counter = 0;
+ headers[counter++] = new RecordHeader(HEADER_DLQ_ERRORS_TOPIC, sourceTopic.getBytes(StandardCharsets.UTF_8));
+ headers[counter++] = new RecordHeader(HEADER_DLQ_ERRORS_PARTITION, Integer.toString(partition).getBytes(StandardCharsets.UTF_8));
+ headers[counter++] = new RecordHeader(HEADER_DLQ_ERRORS_OFFSET, Long.toString(offset).getBytes(StandardCharsets.UTF_8));
+ headers[counter++] = new RecordHeader(HEADER_DLQ_ERRORS_GROUP, groupId.getBytes(StandardCharsets.UTF_8));
+ if (deliveryCount.isPresent()) {
+ headers[counter++] = new RecordHeader(HEADER_DLQ_ERRORS_DELIVERY_COUNT,
+ Short.toString(deliveryCount.get()).getBytes(StandardCharsets.UTF_8));
+ }
+ if (causeMessage != null) {
+ headers[counter] = new RecordHeader(HEADER_DLQ_ERRORS_MESSAGE, causeMessage.getBytes(StandardCharsets.UTF_8));
+ }
+ return headers;
+ }
+
+ /**
+ * Resolves the source topic name from a TopicIdPartition, falling back to the topic ID string.
+ *
+ * @param topicIdPartition The source topic-partition
+ * @param topicNameResolver Resolver that maps topic ID to name
+ * @return The resolved topic name
+ */
+ public static String resolveSourceTopicName(TopicIdPartition topicIdPartition, Function<Uuid, Optional<String>> topicNameResolver) {
+ String recordTopicName = topicIdPartition.topic();
+ if (recordTopicName == null || recordTopicName.isEmpty()) {
+ // If topic name lookup fails, use topic id as a String in the header.
+ recordTopicName = topicNameResolver.apply(topicIdPartition.topicId()).orElse(topicIdPartition.topicId().toString());
+ }
+ return recordTopicName;
+ }
+
+ /**
+ * Computes the destination DLQ partition from the source partition.
+ *
+ * @param sourcePartition The source partition number
+ * @param numDlqPartitions The number of partitions in the DLQ topic
+ * @return The destination DLQ partition number
+ */
+ public static int dlqDestinationPartition(int sourcePartition, int numDlqPartitions) {
+ return sourcePartition % numDlqPartitions;
+ }
+
+ /**
+ * Builds DLQ MemoryRecords for a range of offsets, respecting maxMessageBytes for batch splitting.
+ *
+ * @param param The DLQ record parameter with offset range and metadata
+ * @param resolvedRecordData Map of source offset to source Record (can be incomplete)
+ * @param nextOffsetToSend The first offset to include in this batch
+ * @param lastResolvedOffset The last offset that was resolved by the fetcher
+ * @param maxMessageBytes Maximum batch size in bytes
+ * @param time Time instance for wall-clock timestamps
+ * @param sourceTopic The resolved source topic name
+ * @return BuildResult containing the MemoryRecords and the last offset included
+ */
+ public static BuildResult buildDLQRecords(
+ ShareGroupDLQRecordParameter param,
+ Map<Long, Record> resolvedRecordData,
+ long nextOffsetToSend,
+ long lastResolvedOffset,
+ int maxMessageBytes,
+ Time time,
+ String sourceTopic
+ ) {
+ // In most cases the offset range is a single offset (see DLQ_MAX_FETCH_BYTES). Track the
+ // running batch size incrementally via DefaultRecord.sizeInBytes() - the same formula
+ // MemoryRecords itself uses - instead of re-serializing the whole batch-so-far on every
+ // offset, which would make this loop quadratic in the number of offsets.
+ List<SimpleRecord> simpleRecords = new ArrayList<>();
+ int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
+ Long baseTimestamp = null;
+ // Capped at lastResolvedOffsetThisRound, not just param.lastOffset(): offsets beyond it were
+ // never attempted by this round's fetch and must stay untouched for a fresh round to retry,
+ // rather than being packed in here as headers-only just because they have no map entry yet.
+ // Floored at nextOffsetToSend itself (mirroring the single-record floor below for the
+ // size-exceeds-limit case): a fetch that resolved nothing at all for this round - e.g. a read
+ // that failed outright - would otherwise leave this loop with zero iterations, producing an
+ // empty record batch, which the broker rejects outright. Sending nextOffsetToSend alone,
+ // headers-only, guarantees forward progress even when nothing could be resolved.
+ long roundEnd = Math.max(nextOffsetToSend, Math.min(param.lastOffset(), lastResolvedOffset));
+
+ for (long offset = nextOffsetToSend; offset <= roundEnd; offset++) {
+ // Must be wall-clock (epoch) time: log retention decides whether to delete this
+ // record's segment by comparing its timestamp against the current wall-clock time.
+ long timestamp = time.milliseconds();
+ ByteBuffer key = null;
+ ByteBuffer value = null;
+ Record record = resolvedRecordData.get(offset);
+ if (record != null) {
+ key = record.hasKey() ? record.key() : null;
+ value = record.hasValue() ? record.value() : null;
+ }
+ Header[] recordHeaders = headers(sourceTopic, param.topicIdPartition().partition(),
+ offset, param.groupId(), param.deliveryCount(), param.cause());
+ if (baseTimestamp == null) {
+ baseTimestamp = timestamp;
+ }
+ int recordSize = DefaultRecord.sizeInBytes(simpleRecords.size(), timestamp - baseTimestamp, key, value, recordHeaders);
+
+ if (batchSize + recordSize > maxMessageBytes && !simpleRecords.isEmpty()) {
+ // Adding this record would exceed the limit and the batch already has at least one
+ // record - stop here and send the rest in a follow-up request.
+ break;
+ }
+ simpleRecords.add(new SimpleRecord(timestamp, key, value, recordHeaders));
+ batchSize += recordSize;
+ if (batchSize > maxMessageBytes) {
+ // A single record (with its DLQ headers) already exceeds the limit on its own;
+ // nothing to be gained by holding it back, so send it and let the broker
+ // enforce/report the ultimate limit for this one, rather than stalling forever.
+ break;
+ }
+ }
+
+ long lastOffsetIncluded = nextOffsetToSend + simpleRecords.size() - 1;
+ MemoryRecords records = MemoryRecords.withRecords(
+ Compression.NONE,
+ simpleRecords.toArray(new SimpleRecord[]{})
+ );
+ return new BuildResult(records, lastOffsetIncluded, simpleRecords.size());
+ }
+
+ /**
+ * Optionally fetches source records for DLQ copy-record mode. If copy-record is disabled
+ * for the group, returns an empty result immediately.
+ *
+ * @param param The DLQ record parameter
+ * @param fromOffset The first offset to fetch from
+ * @param cacheHelper Metadata cache helper for config lookups
+ * @param logReader Log reader for fetching source records
+ * @param time Time instance
+ * @param topicNameResolver Resolves the raw DLQ topic name to the name used in the metadata
+ * cache.
+ * @return A future with the fetch result (empty map if copy-record is disabled)
+ */
+ public static CompletableFuture<ShareGroupDLQRecordFetcher.FetchResult> maybeFetchSourceRecords(
+ ShareGroupDLQRecordParameter param,
+ long fromOffset,
+ ShareGroupDLQMetadataCacheHelper cacheHelper,
+ LogReader logReader,
+ Time time,
+ Function<String, String> topicNameResolver
+ ) {
+ if (!cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ return CompletableFuture.completedFuture(
+ new ShareGroupDLQRecordFetcher.FetchResult(Map.of(), param.lastOffset()));
+ }
+
+ // Bounds decompression memory against a pathologically compressible source record: there's
+ // no point retaining more decompressed data than the DLQ topic could ever accept anyway, and
+ // (unlike the user's own topic config) this value isn't controlled by whoever produced the
+ // record being copied. Falls back to DLQ_MAX_FETCH_BYTES in the (defensive-only) case the DLQ
+ // topic isn't resolvable here - copy-record being enabled implies one is configured in practice.
+ int maxDecompressedBytes = cacheHelper.shareGroupDlqTopic(param.groupId())
+ .map(topicNameResolver)
+ .map(cacheHelper::dlqTopicMaxMessageBytes)
+ .orElse(DLQ_MAX_FETCH_BYTES);
+
+ // param itself is never mutated - headers()/topicProduceData() rely on its original,
+ // unwindowed firstOffset/lastOffset for the handler's whole lifetime. Build a throwaway
+ // windowed copy only to scope this round's fetch (and its decompression budget) to what's
+ // left to send.
+ ShareGroupDLQRecordParameter window;
+ if (fromOffset == param.firstOffset()) {
+ window = param;
+ } else {
+ window = new ShareGroupDLQRecordParameter(param.groupId(), param.topicIdPartition(), fromOffset,
+ param.lastOffset(), param.deliveryCount(), param.cause());
+ }
+ return new ShareGroupDLQRecordFetcher(logReader, time, window, DLQ_MAX_FETCH_BYTES, maxDecompressedBytes).fetch();
+ }
+}
diff --git a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
index 0661adca3e8..cc7be522e47 100644
--- a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
+++ b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
@@ -22,13 +22,10 @@ import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.RequestCompletionHandler;
import org.apache.kafka.common.Node;
-import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.TopicConfig;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.message.CreateTopicsRequestData;
import org.apache.kafka.common.message.CreateTopicsResponseData;
import org.apache.kafka.common.message.ProduceRequestData;
@@ -36,8 +33,6 @@ import org.apache.kafka.common.message.ProduceResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.TimestampType;
-import org.apache.kafka.common.record.internal.DefaultRecord;
-import org.apache.kafka.common.record.internal.DefaultRecordBatch;
import org.apache.kafka.common.record.internal.MemoryRecords;
import org.apache.kafka.common.record.internal.Record;
import org.apache.kafka.common.record.internal.SimpleRecord;
@@ -59,8 +54,6 @@ import org.apache.kafka.server.util.timer.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -76,6 +69,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
/**
* Core implementation of RPC send logic for the dlq manager.
@@ -97,13 +91,6 @@ public class ShareGroupDLQStateManager {
private static final int RETRY_BACKOFF_EXP_BASE = CommonClientConfigs.RETRY_BACKOFF_EXP_BASE;
private static final double RETRY_BACKOFF_JITTER = CommonClientConfigs.RETRY_BACKOFF_JITTER;
- /**
- * In most cases we expect the records getting DLQ'ed will be single offsets and
- * not complete batches. Hence, using a large upper limit while reading from the log
- * would be fruitless in most cases. Therefore, the value of 1 MB has been chosen
- * for the DLQ related log reads.
- */
- private static final int DLQ_MAX_FETCH_BYTES = 1024 * 1024;
private static final Logger log = LoggerFactory.getLogger(ShareGroupDLQStateManager.class);
private final Set<Node> inFlight = new HashSet<>();
@@ -307,13 +294,6 @@ public class ShareGroupDLQStateManager {
// so coalesceProduceRequests()'s partition-budget check reuses the exact same value.
private volatile int lastMaxMessageBytes;
- public static final String HEADER_DLQ_ERRORS_TOPIC = "__dlq.errors.topic";
- public static final String HEADER_DLQ_ERRORS_PARTITION = "__dlq.errors.partition";
- public static final String HEADER_DLQ_ERRORS_OFFSET = "__dlq.errors.offset";
- public static final String HEADER_DLQ_ERRORS_GROUP = "__dlq.errors.group";
- public static final String HEADER_DLQ_ERRORS_DELIVERY_COUNT = "__dlq.errors.delivery.count";
- public static final String HEADER_DLQ_ERRORS_MESSAGE = "__dlq.errors.message";
-
public ProduceRequestHandler(
ShareGroupDLQRecordParameter param,
CompletableFuture<Void> result,
@@ -419,7 +399,8 @@ public class ShareGroupDLQStateManager {
throw new ConfigException(String.format("DLQ topic partition leaders for share group %s with DLQ topic %s could not be found.", param.groupId(), dlqTopic.get()));
}
- this.dlqDestinationPartition = param.topicIdPartition().partition() % tpData.numPartitions().get();
+ this.dlqDestinationPartition = ShareGroupDLQRecordHelper.dlqDestinationPartition(
+ param.topicIdPartition().partition(), tpData.numPartitions().get());
this.dlqPartitionLeaderNode = tpData.partitionLeaderNodes().get(dlqDestinationPartition);
if (this.dlqPartitionLeaderNode == null || this.dlqPartitionLeaderNode.equals(Node.noNode())) {
@@ -430,76 +411,24 @@ public class ShareGroupDLQStateManager {
}
public ProduceRequestData.TopicProduceData topicProduceData() {
- // Records have already been resolved (including any remote storage reads) before this
- // handler was added to the node map, so no blocking fetch happens on the sender thread here.
- Map<Long, Record> originalRecordData = resolvedRecordData;
int maxMessageBytes = dlqTopicMaxMessageBytes();
+ String sourceTopic = ShareGroupDLQRecordHelper.resolveSourceTopicName(
+ param.topicIdPartition(), topicId -> cacheHelper.topicName(topicId));
- // In most cases the offset range is a single offset (see DLQ_MAX_FETCH_BYTES). Track the
- // running batch size incrementally via DefaultRecord.sizeInBytes() - the same formula
- // MemoryRecords itself uses - instead of re-serializing the whole batch-so-far on every
- // offset, which would make this loop quadratic in the number of offsets.
- List<SimpleRecord> simpleRecords = new ArrayList<>();
- int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
- Long baseTimestamp = null;
- // Capped at lastResolvedOffsetThisRound, not just param.lastOffset(): offsets beyond it were
- // never attempted by this round's fetch and must stay untouched for a fresh round to retry,
- // rather than being packed in here as headers-only just because they have no map entry yet.
- // Floored at nextOffsetToSend itself (mirroring the single-record floor below for the
- // size-exceeds-limit case): a fetch that resolved nothing at all for this round - e.g. a read
- // that failed outright - would otherwise leave this loop with zero iterations, producing an
- // empty record batch, which the broker rejects outright. Sending nextOffsetToSend alone,
- // headers-only, guarantees forward progress even when nothing could be resolved.
- long roundEnd = Math.max(nextOffsetToSend, Math.min(param.lastOffset(), lastResolvedOffsetThisRound));
- for (long offset = nextOffsetToSend; offset <= roundEnd; offset++) {
- // Must be wall-clock (epoch) time: log retention decides whether to delete this
- // record's segment by comparing its timestamp against the current wall-clock time.
- long timestamp = time.milliseconds();
- ByteBuffer key = null;
- ByteBuffer value = null;
- Record record = originalRecordData.get(offset);
- if (record != null) {
- key = record.hasKey() ? record.key() : null;
- value = record.hasValue() ? record.value() : null;
- }
- Header[] recordHeaders = headers(offset);
- if (baseTimestamp == null) {
- baseTimestamp = timestamp;
- }
- int recordSize = DefaultRecord.sizeInBytes(simpleRecords.size(), timestamp - baseTimestamp, key, value, recordHeaders);
+ ShareGroupDLQRecordHelper.BuildResult buildResult = ShareGroupDLQRecordHelper.buildDLQRecords(
+ param, resolvedRecordData, nextOffsetToSend, lastResolvedOffsetThisRound,
+ maxMessageBytes, time, sourceTopic);
- if (batchSize + recordSize > maxMessageBytes && !simpleRecords.isEmpty()) {
- // Adding this record would exceed the limit and the batch already has at least one
- // record - stop here and send the rest in a follow-up request.
- break;
- }
- simpleRecords.add(new SimpleRecord(timestamp, key, value, recordHeaders));
- batchSize += recordSize;
- if (batchSize > maxMessageBytes) {
- // A single record (with its DLQ headers) already exceeds the limit on its own;
- // nothing to be gained by holding it back, so send it and let the broker
- // enforce/report the ultimate limit for this one, rather than stalling forever.
- break;
- }
- }
- lastOffsetIncludedThisRound = nextOffsetToSend + simpleRecords.size() - 1;
- // Cached so coalesceProduceRequests()'s partition-budget check can reuse the exact value
- // used above to build this batch, instead of looking it up (and dynamically re-resolving
- // it) a second time for the same round.
+ lastOffsetIncludedThisRound = buildResult.lastOffsetIncluded();
this.lastMaxMessageBytes = maxMessageBytes;
- MemoryRecords records = MemoryRecords.withRecords(
- Compression.NONE,
- simpleRecords.toArray(new SimpleRecord[]{})
- );
-
return new ProduceRequestData.TopicProduceData()
.setName(dlqTopicPartitionData.topicName())
.setTopicId(dlqTopicPartitionData.topicId().get())
.setPartitionData(List.of(
new ProduceRequestData.PartitionProduceData()
.setIndex(dlqDestinationPartition) // partition
- .setRecords(records)
+ .setRecords(buildResult.records())
));
}
@@ -521,20 +450,18 @@ public class ShareGroupDLQStateManager {
public Optional<Throwable> validateDlqTopic() {
Optional<String> topicNameOpt = cacheHelper.shareGroupDlqTopic(param.groupId());
- Optional<String> topicPrefix = cacheHelper.shareGroupDlqTopicPrefix();
// Verify that DLQ topic for the share group is set and is correctly named.
if (topicNameOpt.isEmpty()) {
return Optional.of(new ConfigException(String.format("Configured DLQ topic name in share group: %s is empty.", param.groupId())));
- } else if (topicNameOpt.get().startsWith("__")) {
- return Optional.of(new ConfigException(String.format("Configured DLQ topic name in share group: %s cannot start with __, topic: %s.", param.groupId(), topicNameOpt.get())));
}
String topicName = topicNameOpt.get();
- // Verify that DLQ is enabled on a correctly named topic, configured on a share group.
- if (cacheHelper.containsTopic(topicName) && !cacheHelper.isDlqEnabledOnTopic(topicName)) {
- return Optional.of(new ConfigException(String.format("DLQ is not enabled on configured DLQ topic for share group: %s, topic: %s.", param.groupId(), topicName)));
+ Optional<Throwable> sharedError = ShareGroupDLQValidator.validateDlqTopicConfig(
+ param.groupId(), topicName, topicName, cacheHelper);
+ if (sharedError.isPresent()) {
+ return sharedError;
}
// Verify that for a non-existent correctly named DLQ topic, auto create should be enabled.
@@ -542,13 +469,7 @@ public class ShareGroupDLQStateManager {
return Optional.of(new ConfigException(String.format("DLQ topic does not exist and auto create is disabled on cluster for share group: %s, topic: %s.", param.groupId(), topicName)));
}
- // Verify that if configured, the DLQ topic name prefix aligns with the topic name.
- return topicPrefix.map(prefix -> {
- if (!prefix.isEmpty() && !topicName.startsWith(prefix)) {
- return new ConfigException(String.format("Configured DLQ topic name does not comply with the DLQ topic prefix in share group: %s, topic: %s, prefix: %s.", param.groupId(), topicName, prefix));
- }
- return null;
- });
+ return Optional.empty();
}
public boolean dlqTopicExists() {
@@ -578,32 +499,6 @@ public class ShareGroupDLQStateManager {
")";
}
- private Header[] headers(long offset) {
- List<Header> headers = new ArrayList<>();
- headers.add(new RecordHeader(HEADER_DLQ_ERRORS_TOPIC, recordTopic().getBytes(StandardCharsets.UTF_8)));
- headers.add(new RecordHeader(HEADER_DLQ_ERRORS_PARTITION, Integer.toString(param.topicIdPartition().partition()).getBytes(StandardCharsets.UTF_8)));
- headers.add(new RecordHeader(HEADER_DLQ_ERRORS_OFFSET, Long.toString(offset).getBytes(StandardCharsets.UTF_8)));
- headers.add(new RecordHeader(HEADER_DLQ_ERRORS_GROUP, param.groupId().getBytes(StandardCharsets.UTF_8)));
- param.deliveryCount().ifPresent(deliveryCount -> headers.add(
- new RecordHeader(HEADER_DLQ_ERRORS_DELIVERY_COUNT, Short.toString(deliveryCount).getBytes(StandardCharsets.UTF_8))));
- param.cause().ifPresent(cause -> {
- if (cause.getMessage() != null) {
- headers.add(new RecordHeader(HEADER_DLQ_ERRORS_MESSAGE, cause.getMessage().getBytes(StandardCharsets.UTF_8)));
- }
- });
-
- return headers.toArray(new Header[0]);
- }
-
- private String recordTopic() {
- TopicIdPartition topicIdPartition = param.topicIdPartition();
- String recordTopicName = param.topicIdPartition().topic();
- if (recordTopicName == null || recordTopicName.isEmpty()) {
- // If topic name lookup fails, use topic id as a String in the header.
- recordTopicName = cacheHelper.topicName(param.topicIdPartition().topicId()).orElse(topicIdPartition.topicId().toString());
- }
- return recordTopicName;
- }
// Visibility for testing
Optional<Errors> checkResponseError(ClientResponse response) {
@@ -866,30 +761,8 @@ public class ShareGroupDLQStateManager {
}
private CompletableFuture<ShareGroupDLQRecordFetcher.FetchResult> maybeFetchRecordData(long fromOffset) {
- if (!cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
- return CompletableFuture.completedFuture(
- new ShareGroupDLQRecordFetcher.FetchResult(Map.of(), param.lastOffset()));
- }
- // Bounds decompression memory against a pathologically compressible source record: there's
- // no point retaining more decompressed data than the DLQ topic could ever accept anyway, and
- // (unlike the user's own topic config) this value isn't controlled by whoever produced the
- // record being copied. Falls back to DLQ_MAX_FETCH_BYTES in the (defensive-only) case the DLQ
- // topic isn't resolvable here - copy-record being enabled implies one is configured in practice.
- int maxDecompressedBytes = cacheHelper.shareGroupDlqTopic(param.groupId())
- .map(cacheHelper::dlqTopicMaxMessageBytes)
- .orElse(DLQ_MAX_FETCH_BYTES);
- // param itself is never mutated - headers()/topicProduceData() rely on its original,
- // unwindowed firstOffset/lastOffset for the handler's whole lifetime. Build a throwaway
- // windowed copy only to scope this round's fetch (and its decompression budget) to what's
- // left to send.
- ShareGroupDLQRecordParameter window;
- if (fromOffset == param.firstOffset()) {
- window = param;
- } else {
- window = new ShareGroupDLQRecordParameter(param.groupId(), param.topicIdPartition(), fromOffset,
- param.lastOffset(), param.deliveryCount(), param.cause());
- }
- return new ShareGroupDLQRecordFetcher(logReader, time, window, DLQ_MAX_FETCH_BYTES, maxDecompressedBytes).fetch();
+ return ShareGroupDLQRecordHelper.maybeFetchSourceRecords(
+ param, fromOffset, cacheHelper, logReader, time, Function.identity());
}
}
diff --git a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQValidator.java b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQValidator.java
new file mode 100644
index 00000000000..1a5ff10616f
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQValidator.java
@@ -0,0 +1,106 @@
+/*
+ * 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.kafka.server.share.dlq;
+
+import org.apache.kafka.common.config.ConfigException;
+
+import java.util.Optional;
+
+/**
+ * Validation logic for DLQ manager.
+ */
+public final class ShareGroupDLQValidator {
+
+ private ShareGroupDLQValidator() {}
+
+ /**
+ * Validates the fields of a {@link ShareGroupDLQRecordParameter}.
+ *
+ * @throws IllegalArgumentException if any field is invalid
+ */
+ public static void validateParam(ShareGroupDLQRecordParameter param) {
+ String prefix = "DLQ records parameters";
+ if (param == null) {
+ throw new IllegalArgumentException(prefix + " cannot be null.");
+ }
+ if (param.groupId() == null || param.groupId().isEmpty()) {
+ throw new IllegalArgumentException(prefix + " group cannot be null or empty.");
+ }
+ if (param.topicIdPartition() == null) {
+ throw new IllegalArgumentException(prefix + " topic/partition data cannot be null or empty.");
+ }
+ if (param.topicIdPartition().topicId() == null) {
+ throw new IllegalArgumentException(prefix + " topic id data cannot be null or empty.");
+ }
+ if (param.topicIdPartition().partition() < 0) {
+ throw new IllegalArgumentException(prefix + " partition cannot be negative.");
+ }
+ if (param.firstOffset() < 0) {
+ throw new IllegalArgumentException(prefix + " first offset cannot be negative.");
+ }
+ if (param.lastOffset() < 0) {
+ throw new IllegalArgumentException(prefix + " last offset cannot be negative.");
+ }
+ if (param.lastOffset() < param.firstOffset()) {
+ throw new IllegalArgumentException(prefix + " last offset cannot be less than first offset.");
+ }
+ }
+
+ /**
+ * Validates DLQ topic configuration. Checks that the topic name does not start with {@code __},
+ * that DLQ is enabled on the topic (if it exists), and that the topic name complies with the
+ * configured prefix.
+ *
+ * <p>Callers are responsible for checking that the topic name is present in the config (non-empty)
+ * before calling this method, and for any implementation-specific checks (e.g., auto-create).
+ *
+ * @param groupId the share group ID, for error messages
+ * @param userTopicName the raw DLQ topic name from config
+ * @param resolvedTopicName the topic name used for metadata cache lookups
+ * @param cacheHelper metadata cache helper for topic lookups
+ * @return an error if validation fails, or empty if valid
+ */
+ public static Optional<Throwable> validateDlqTopicConfig(
+ String groupId,
+ String userTopicName,
+ String resolvedTopicName,
+ ShareGroupDLQMetadataCacheHelper cacheHelper
+ ) {
+ if (userTopicName.startsWith("__")) {
+ return Optional.of(new ConfigException(String.format(
+ "Configured DLQ topic name in share group: %s cannot start with __, topic: %s.", groupId, userTopicName)));
+ }
+
+ // Verify that DLQ is enabled on a correctly named topic, configured on a share group.
+ if (cacheHelper.containsTopic(resolvedTopicName) && !cacheHelper.isDlqEnabledOnTopic(resolvedTopicName)) {
+ return Optional.of(new ConfigException(
+ "DLQ is not enabled on configured DLQ topic for share group: "
+ + groupId + ", topic: " + userTopicName));
+ }
+
+ Optional<String> topicPrefix = cacheHelper.shareGroupDlqTopicPrefix();
+ return topicPrefix.map(prefix -> {
+ if (!prefix.isEmpty() && !userTopicName.startsWith(prefix)) {
+ return new ConfigException(
+ "Configured DLQ topic name does not comply with the DLQ topic prefix in share group: "
+ + groupId + ", topic: " + userTopicName + ", prefix: " + prefix);
+ }
+ return null;
+ });
+ }
+}
diff --git a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
index 5e88c3f9dbc..441d0b92c52 100644
--- a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
+++ b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
@@ -80,12 +80,12 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
-import static org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.ProduceRequestHandler.HEADER_DLQ_ERRORS_DELIVERY_COUNT;
-import static org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.ProduceRequestHandler.HEADER_DLQ_ERRORS_GROUP;
-import static org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.ProduceRequestHandler.HEADER_DLQ_ERRORS_MESSAGE;
-import static org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.ProduceRequestHandler.HEADER_DLQ_ERRORS_OFFSET;
-import static org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.ProduceRequestHandler.HEADER_DLQ_ERRORS_PARTITION;
-import static org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.ProduceRequestHandler.HEADER_DLQ_ERRORS_TOPIC;
+import static org.apache.kafka.server.share.dlq.ShareGroupDLQRecordHelper.HEADER_DLQ_ERRORS_DELIVERY_COUNT;
+import static org.apache.kafka.server.share.dlq.ShareGroupDLQRecordHelper.HEADER_DLQ_ERRORS_GROUP;
+import static org.apache.kafka.server.share.dlq.ShareGroupDLQRecordHelper.HEADER_DLQ_ERRORS_MESSAGE;
+import static org.apache.kafka.server.share.dlq.ShareGroupDLQRecordHelper.HEADER_DLQ_ERRORS_OFFSET;
+import static org.apache.kafka.server.share.dlq.ShareGroupDLQRecordHelper.HEADER_DLQ_ERRORS_PARTITION;
+import static org.apache.kafka.server.share.dlq.ShareGroupDLQRecordHelper.HEADER_DLQ_ERRORS_TOPIC;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;