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 83947cc6452 KAFKA-20771: Enforce max msg bytes in share group DLQ
produce path. (#22802)
83947cc6452 is described below
commit 83947cc6452f215ee0c590b8ddb299dec2a7a393
Author: Sushant Mahajan <[email protected]>
AuthorDate: Mon Jul 13 20:10:21 2026 +0530
KAFKA-20771: Enforce max msg bytes in share group DLQ produce path. (#22802)
- Add `dlqTopicMaxMessageBytes(topic`) to
`ShareGroupDLQMetadataCacheHelper`,
honoring a topic-level max.message.bytes override or else falling back
to the broker's dynamically-reconfigurable message.max.bytes (via
`IntSupplier`, not a captured value).
- `ShareGroupDLQStateManager` now chunks
an oversized DLQ offset range across multiple produce requests with a
resumable per-handler integer, and defers (rather than fails) coalesced
handlers that would push a merged batch over the limit, re-enqueuing
them for a later round instead of sending oversized requests.
- Also, switches the per-record size check to an incremental calculation
(`DefaultRecord.sizeInBytes`) instead of re-serializing the batch-so-far
on every offset.
- Pins `message.timestamp.type=CreateTime` on
auto-created DLQ topics so timestamps aren't silently overwritten by
the cluster's broker-level default.
- Add new tests and update existing wherever applicable.
- Add integ test to verify the code path.
Reviewers: Apoorv Mittal <[email protected]>
---
.../clients/consumer/ShareConsumerDLQTest.java | 137 ++++++++++++++++
.../ShareCoordinatorMetadataCacheHelperImpl.java | 22 ++-
.../src/main/scala/kafka/server/BrokerServer.scala | 4 +-
...hareCoordinatorMetadataCacheHelperImplTest.java | 180 +++++++++++++++++----
.../dlq/ShareGroupDLQMetadataCacheHelper.java | 10 ++
.../share/dlq/ShareGroupDLQStateManager.java | 139 ++++++++++++++--
.../share/dlq/ShareGroupDLQStateManagerTest.java | 126 +++++++++++++++
7 files changed, 565 insertions(+), 53 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 1cbf5d16e02..2fe897fdd71 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
@@ -17,6 +17,8 @@
package org.apache.kafka.clients.consumer;
import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.NewTopic;
@@ -48,6 +50,7 @@ import java.util.Set;
import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
import static org.apache.kafka.test.TestUtils.waitForCondition;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -462,6 +465,140 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
verifyDlqMetrics(groupId, recordCount);
}
+ /**
+ * Verifies that the DLQ topic's own {@code max.message.bytes} - not the
source topic's - bounds each DLQ
+ * produce request, and that this is honored dynamically when the config
changes.
+ *
+ * <p>Phase 1: the source topic allows records up to {@code
sourceMaxMessageBytes}, but the DLQ topic is
+ * configured with a third of that ({@code dlqMaxMessageBytes}). Three
records, each sized so that any two
+ * of their (record-copy-enabled) DLQ copies together would exceed
dlqMaxMessageBytes, are rejected together
+ * in a single commit - so SharePartition issues one DLQ call spanning the
whole offset range (see
+ * ShareGroupDLQStateManagerTest for the underlying chunking logic). The
DLQ produce path must then split
+ * that single call into multiple sequential produce requests (one record
each) rather than failing or
+ * dropping any record; this confirms all 3 records still land on the DLQ
topic, and that the DLQ
+ * produce-request count increased by at least 3 (one per chunk) -
concrete proof splitting occurred, since
+ * without it the single oversized request would be rejected once
(non-retriable) and no records would ever
+ * arrive.
+ *
+ * <p>Phase 2: the DLQ topic's {@code max.message.bytes} is then raised
via {@code IncrementalAlterConfigs}
+ * to comfortably exceed what a fresh batch of 3 more (same-sized) records
needs combined, and the reject
+ * scenario is repeated with that new batch. This confirms two more
things: the broker picks up the raised
+ * limit dynamically (not a value cached at startup), and the chunking
logic does not split unnecessarily
+ * once the budget is actually sufficient - the produce-request count must
increase by exactly 1 for the
+ * second batch, not 3.
+ */
+ @ClusterTest
+ public void testDlqRespectsDlqTopicMaxMessageBytesNotEqToSourceTopic()
throws Exception {
+ String groupId = "dlq-maxbytes-group";
+ String sourceTopic = "dlq-maxbytes-source";
+ String dlqTopic = "dlq.maxbytes";
+ int recordCount = 3;
+ int sourceMaxMessageBytes = 300_000;
+ int dlqMaxMessageBytes = sourceMaxMessageBytes / 3;
+ // Leave headroom below dlqMaxMessageBytes for the DLQ context
headers/record-batch framing overhead,
+ // so a single record's DLQ copy cleanly fits under the limit but two
together clearly don't.
+ int payloadSize = dlqMaxMessageBytes - 2_000;
+
+ try (Admin admin = createAdminClient()) {
+ admin.createTopics(Set.of(
+ new NewTopic(sourceTopic, 1, (short) 1)
+ .configs(Map.of(TopicConfig.MAX_MESSAGE_BYTES_CONFIG,
Integer.toString(sourceMaxMessageBytes))),
+ new NewTopic(dlqTopic, 1, (short) 1)
+ .configs(Map.of(
+
TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG, "true",
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG,
Integer.toString(dlqMaxMessageBytes)))
+ )).all().get();
+ }
+
+ alterShareAutoOffsetReset(groupId, "earliest");
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+ // Record copy must be enabled - otherwise DLQ records carry headers
only (tiny) and would never
+ // approach dlqMaxMessageBytes regardless of the source record size.
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_COPY_RECORD_ENABLE_CONFIG, "true");
+
+ byte[] payload = new byte[payloadSize];
+ 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));
+ }
+ producer.flush();
+ }
+
+ // Reject all 3 records together in one commit, so SharePartition
issues a single DLQ call spanning
+ // the whole offset range (a fresh, contiguous, single-fetch
acquisition with no prior redeliveries
+ // maps to one cached in-flight batch, so one client-side reject
commit produces one DLQ call).
+ rejectRecords(groupId, sourceTopic, recordCount);
+
+ // All 3 records must still reach the DLQ, split across multiple
produce requests since no pairing
+ // of their DLQ copies fits within dlqMaxMessageBytes. Verified inline
(rather than via
+ // verifyDlqTopicRecords()) since that helper hardcodes checking the
copied value against the fixed
+ // "value" content produced by produceMessages()/produceTo(), not this
test's large payload.
+ List<ConsumerRecord<byte[], byte[]>> dlqRecords =
readDlqPartition(dlqTopic, 0, recordCount);
+ assertEquals(recordCount, dlqRecords.size(), "Unexpected number of
records on the DLQ topic");
+ Set<Long> actualSourceOffsets = new HashSet<>();
+ for (ConsumerRecord<byte[], byte[]> record : dlqRecords) {
+ assertArrayEquals(payload, record.value(), "DLQ record value
should be the copied payload");
+ assertEquals(groupId, headerValue(record,
HEADER_DLQ_ERRORS_GROUP));
+ assertEquals(sourceTopic, headerValue(record,
HEADER_DLQ_ERRORS_TOPIC));
+ assertEquals("0", headerValue(record,
HEADER_DLQ_ERRORS_PARTITION));
+
actualSourceOffsets.add(Long.parseLong(Objects.requireNonNull(headerValue(record,
HEADER_DLQ_ERRORS_OFFSET))));
+ }
+ assertEquals(expectedSourceOffsets(recordCount), actualSourceOffsets,
+ "DLQ records should cover every expected source offset");
+ verifyDlqMetrics(groupId, recordCount);
+
+ // Concrete proof that splitting - not some other mechanism - is why
all 3 records arrived: there is
+ // exactly one logical DLQ call here (one contiguous reject batch,
offsets 0-2), so any produce-request
+ // count above 1 for this group can only come from the
resumable-cursor chunking logic splitting that
+ // one call into multiple sequential produce requests to stay within
dlqMaxMessageBytes. Without it,
+ // the single oversized request would be rejected once
(MESSAGE_TOO_LARGE is not retriable) and no
+ // records would ever reach the DLQ - contradicting the assertions
above. >= rather than == tolerates
+ // an occasional extra retry (e.g. a transient network blip) without
being flaky.
+ assertTrue(dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL, groupId) >=
recordCount,
+ "Expected at least " + recordCount + " separate DLQ produce
requests (one per chunk), was "
+ + dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL, groupId));
+ long produceCountBeforeRaise = dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL,
groupId);
+
+ // Now raise the DLQ topic's max.message.bytes well above what all 3
(record-copy-enabled) DLQ copies
+ // need together, and repeat the same reject scenario with a fresh
batch of 3 records. This confirms
+ // two things at once: dlqTopicMaxMessageBytes() picks up the change
dynamically (it wraps a live
+ // topic-config lookup, not a value captured once at startup - see
ShareCoordinatorMetadataCacheHelperImpl),
+ // and the chunking logic doesn't split unnecessarily when the budget
is actually sufficient - the
+ // produce-request count must increase by exactly 1 (one request for
the whole new batch), not 3.
+ int raisedDlqMaxMessageBytes = sourceMaxMessageBytes * 2;
+ ConfigResource dlqTopicResource = new
ConfigResource(ConfigResource.Type.TOPIC, dlqTopic);
+ try (Admin admin = createAdminClient()) {
+ admin.incrementalAlterConfigs(
+ Map.of(dlqTopicResource, List.of(new AlterConfigOp(
+ new ConfigEntry(TopicConfig.MAX_MESSAGE_BYTES_CONFIG,
Integer.toString(raisedDlqMaxMessageBytes)),
+ AlterConfigOp.OpType.SET))),
+ new AlterConfigsOptions()
+ ).all().get();
+ waitForCondition(() -> {
+ Config config =
admin.describeConfigs(List.of(dlqTopicResource)).all().get().get(dlqTopicResource);
+ ConfigEntry entry =
config.get(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+ return entry != null &&
entry.value().equals(Integer.toString(raisedDlqMaxMessageBytes));
+ }, DEFAULT_MAX_WAIT_MS, 100L, () -> "Raised max.message.bytes did
not propagate on the DLQ topic");
+ }
+
+ 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));
+ }
+ producer.flush();
+ }
+ rejectRecords(groupId, sourceTopic, recordCount);
+
+ List<ConsumerRecord<byte[], byte[]>> secondBatchDlqRecords =
readDlqPartition(dlqTopic, 0, recordCount * 2);
+ assertEquals(recordCount * 2, secondBatchDlqRecords.size(),
+ "Unexpected number of records on the DLQ topic after the second
batch");
+
+ waitForCondition(() -> dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL,
groupId) == produceCountBeforeRaise + 1,
+ DEFAULT_MAX_WAIT_MS, 200L,
+ () -> "Expected exactly 1 additional DLQ produce request for the
second batch (budget no longer forces "
+ + "chunking), count went from " + produceCountBeforeRaise + "
to " + dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL, groupId));
+ }
+
/**
* Verifies the DLQ is NOT written when it is gated off, across the gating
conditions:
* (a) no DLQ topic configured for the group;
diff --git
a/core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java
b/core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java
index 6e0667cc6a9..8d1fce822a2 100644
---
a/core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java
+++
b/core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java
@@ -44,24 +44,28 @@ import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.function.Function;
+import java.util.function.IntSupplier;
public class ShareCoordinatorMetadataCacheHelperImpl implements
ShareCoordinatorMetadataCacheHelper, ShareGroupDLQMetadataCacheHelper {
private final MetadataCache metadataCache;
private final Function<SharePartitionKey, Integer> keyToPartitionMapper;
private final ListenerName interBrokerListenerName;
private final GroupConfigManager groupConfigManager;
+ private final IntSupplier messageMaxBytesSupplier;
private final Logger log =
LoggerFactory.getLogger(ShareCoordinatorMetadataCacheHelperImpl.class);
public ShareCoordinatorMetadataCacheHelperImpl(
MetadataCache metadataCache,
Function<SharePartitionKey, Integer> keyToPartitionMapper,
ListenerName interBrokerListenerName,
- GroupConfigManager groupConfigManager
+ GroupConfigManager groupConfigManager,
+ IntSupplier messageMaxBytesSupplier
) {
this.metadataCache = Objects.requireNonNull(metadataCache,
"metadataCache must not be null");
this.keyToPartitionMapper =
Objects.requireNonNull(keyToPartitionMapper, "keyToPartitionMapper must not be
null");
this.interBrokerListenerName =
Objects.requireNonNull(interBrokerListenerName, "interBrokerListenerName must
not be null");
this.groupConfigManager = Objects.requireNonNull(groupConfigManager,
"groupConfigManager must not be null");
+ this.messageMaxBytesSupplier =
Objects.requireNonNull(messageMaxBytesSupplier, "messageMaxBytesSupplier must
not be null");
}
@Override
@@ -103,6 +107,22 @@ public class ShareCoordinatorMetadataCacheHelperImpl
implements ShareCoordinator
}
}
+ @Override
+ public int dlqTopicMaxMessageBytes(String topic) {
+ Properties props = metadataCache.topicConfig(topic);
+ // LogConfig defines its own static default for this key, so an
explicit containsKey
+ // check is needed to distinguish "no topic-level override" from
"override present"
+ // and fall back to the broker's configured message.max.bytes in the
former case.
+ if (props == null ||
!props.containsKey(TopicConfig.MAX_MESSAGE_BYTES_CONFIG)) {
+ return messageMaxBytesSupplier.getAsInt();
+ }
+ try {
+ return new
LogConfig(props).getInt(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+ } catch (ConfigException exe) {
+ return messageMaxBytesSupplier.getAsInt();
+ }
+ }
+
@Override
public boolean isShareGroupDlqCopyRecordEnabled(String groupId) {
Optional<GroupConfig> groupConfig =
groupConfigManager.groupConfig(groupId);
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala
b/core/src/main/scala/kafka/server/BrokerServer.scala
index d7502aaa039..f7f34c17537 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -755,7 +755,7 @@ class BrokerServer(
if (klass.getName.equals(classOf[DefaultStatePersister].getName)) {
DefaultStatePersister.instance(
NetworkUtils.buildNetworkClient("Persister", config, metrics,
Time.SYSTEM, new LogContext(s"[Persister broker=${config.brokerId}]")),
- new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key =>
shareCoordinator.partitionFor(key), config.interBrokerListenerName,
groupConfigManager),
+ new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key =>
shareCoordinator.partitionFor(key), config.interBrokerListenerName,
groupConfigManager, () => config.messageMaxBytes),
Time.SYSTEM,
shareGroupTimer
)
@@ -779,7 +779,7 @@ class BrokerServer(
if (klass.getName.equals(classOf[DefaultShareGroupDLQManager].getName)) {
DefaultShareGroupDLQManager.instance(
NetworkUtils.buildNetworkClient("ShareGroupDLQManager", config,
metrics, Time.SYSTEM, new LogContext(s"[ShareGroupDLQManager
broker=${config.brokerId}]")),
- new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key =>
shareCoordinator.partitionFor(key), config.interBrokerListenerName,
groupConfigManager),
+ new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key =>
shareCoordinator.partitionFor(key), config.interBrokerListenerName,
groupConfigManager, () => config.messageMaxBytes),
Time.SYSTEM,
shareGroupTimer,
shareGroupMetrics,
diff --git
a/core/src/test/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImplTest.java
b/core/src/test/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImplTest.java
index 713a4d78909..0ca8717a293 100644
---
a/core/src/test/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImplTest.java
+++
b/core/src/test/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImplTest.java
@@ -28,6 +28,7 @@ import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.coordinator.group.GroupConfig;
import org.apache.kafka.coordinator.group.GroupConfigManager;
import org.apache.kafka.metadata.MetadataCache;
+import org.apache.kafka.server.config.ServerLogConfigs;
import org.apache.kafka.server.share.SharePartitionKey;
import org.apache.kafka.server.share.dlq.ShareGroupDLQMetadataCacheHelper;
import
org.apache.kafka.server.share.persister.ShareCoordinatorMetadataCacheHelper;
@@ -60,7 +61,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
null,
func,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
));
assertEquals("metadataCache must not be null", e.getMessage());
@@ -68,7 +70,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
null,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
));
assertEquals("keyToPartitionMapper must not be null", e.getMessage());
@@ -76,7 +79,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
func,
null,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
));
assertEquals("interBrokerListenerName must not be null",
e.getMessage());
@@ -84,9 +88,19 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
func,
mock(ListenerName.class),
- null
+ null,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
));
assertEquals("groupConfigManager must not be null", e.getMessage());
+
+ e = assertThrows(NullPointerException.class, () -> new
ShareCoordinatorMetadataCacheHelperImpl(
+ mock(MetadataCache.class),
+ func,
+ mock(ListenerName.class),
+ mock(GroupConfigManager.class),
+ null
+ ));
+ assertEquals("messageMaxBytesSupplier must not be null",
e.getMessage());
}
@Test
@@ -100,7 +114,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
func,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
when(mockMetadataCache.contains(eq(Topic.SHARE_GROUP_STATE_TOPIC_NAME)))
@@ -123,7 +138,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
func,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
when(mockMetadataCache.contains(eq(Topic.SHARE_GROUP_STATE_TOPIC_NAME)))
@@ -146,7 +162,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
func,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertEquals(
@@ -167,7 +184,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
func,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
when(mockMetadataCache.contains(eq(Topic.SHARE_GROUP_STATE_TOPIC_NAME)))
@@ -256,7 +274,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
func,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
when(mockMetadataCache.contains(eq(Topic.SHARE_GROUP_STATE_TOPIC_NAME)))
@@ -313,7 +332,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
func,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
when(mockMetadataCache.contains(eq(Topic.SHARE_GROUP_STATE_TOPIC_NAME)))
@@ -370,7 +390,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
func,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
when(mockMetadataCache.contains(eq(Topic.SHARE_GROUP_STATE_TOPIC_NAME)))
@@ -428,7 +449,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
func,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
when(mockMetadataCache.getAliveBrokerNodes(
@@ -455,7 +477,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
func,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
List<Node> nodes = List.of(
@@ -488,7 +511,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertEquals(Optional.empty(), cache.shareGroupDlqTopic("test-group"));
@@ -506,7 +530,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertEquals(Optional.of("dlq-topic"),
cache.shareGroupDlqTopic("test-group"));
@@ -524,7 +549,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertFalse(cache.isShareGroupDlqCopyRecordEnabled("test-group"));
@@ -542,7 +568,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertTrue(cache.isShareGroupDlqCopyRecordEnabled("test-group"));
@@ -560,7 +587,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertFalse(cache.isShareGroupDlqCopyRecordEnabled("test-group"));
@@ -578,7 +606,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertTrue(cache.isDlqAutoTopicCreateEnabled());
@@ -594,7 +623,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertFalse(cache.isDlqAutoTopicCreateEnabled());
@@ -612,7 +642,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertEquals(Optional.of("dlq."), cache.shareGroupDlqTopicPrefix());
@@ -628,7 +659,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertEquals(Optional.empty(), cache.shareGroupDlqTopicPrefix());
@@ -646,7 +678,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertFalse(cache.isDlqEnabledOnTopic("test-topic"));
@@ -662,7 +695,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertFalse(cache.isDlqEnabledOnTopic("test-topic"));
@@ -679,7 +713,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertTrue(cache.isDlqEnabledOnTopic("test-topic"));
@@ -696,12 +731,84 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertFalse(cache.isDlqEnabledOnTopic("test-topic"));
}
+ // Tests for dlqTopicMaxMessageBytes
+
+ @Test
+ public void
testDlqTopicMaxMessageBytesFallsBackToBrokerConfigWhenTopicConfigNull() {
+ MetadataCache mockMetadataCache = mock(MetadataCache.class);
+ when(mockMetadataCache.topicConfig("test-topic")).thenReturn(null);
+
+ ShareCoordinatorMetadataCacheHelperImpl cache = new
ShareCoordinatorMetadataCacheHelperImpl(
+ mockMetadataCache,
+ sharePartitionKey -> 0,
+ mock(ListenerName.class),
+ mock(GroupConfigManager.class),
+ () -> 5_000_000
+ );
+
+ assertEquals(5_000_000, cache.dlqTopicMaxMessageBytes("test-topic"));
+ }
+
+ @Test
+ public void
testDlqTopicMaxMessageBytesFallsBackToBrokerConfigWhenNoTopicOverride() {
+ MetadataCache mockMetadataCache = mock(MetadataCache.class);
+ when(mockMetadataCache.topicConfig("test-topic")).thenReturn(new
Properties());
+
+ ShareCoordinatorMetadataCacheHelperImpl cache = new
ShareCoordinatorMetadataCacheHelperImpl(
+ mockMetadataCache,
+ sharePartitionKey -> 0,
+ mock(ListenerName.class),
+ mock(GroupConfigManager.class),
+ () -> 5_000_000
+ );
+
+ assertEquals(5_000_000, cache.dlqTopicMaxMessageBytes("test-topic"));
+ }
+
+ @Test
+ public void testDlqTopicMaxMessageBytesUsesTopicOverrideWhenPresent() {
+ MetadataCache mockMetadataCache = mock(MetadataCache.class);
+ Properties props = new Properties();
+ props.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, "10000000");
+ when(mockMetadataCache.topicConfig("test-topic")).thenReturn(props);
+
+ ShareCoordinatorMetadataCacheHelperImpl cache = new
ShareCoordinatorMetadataCacheHelperImpl(
+ mockMetadataCache,
+ sharePartitionKey -> 0,
+ mock(ListenerName.class),
+ mock(GroupConfigManager.class),
+ () -> 5_000_000
+ );
+
+ // Topic-level override wins even though it is larger than the
broker-configured default.
+ assertEquals(10_000_000, cache.dlqTopicMaxMessageBytes("test-topic"));
+ }
+
+ @Test
+ public void
testDlqTopicMaxMessageBytesFallsBackToBrokerConfigOnInvalidTopicOverride() {
+ MetadataCache mockMetadataCache = mock(MetadataCache.class);
+ Properties props = new Properties();
+ props.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, "not-a-number");
+ when(mockMetadataCache.topicConfig("test-topic")).thenReturn(props);
+
+ ShareCoordinatorMetadataCacheHelperImpl cache = new
ShareCoordinatorMetadataCacheHelperImpl(
+ mockMetadataCache,
+ sharePartitionKey -> 0,
+ mock(ListenerName.class),
+ mock(GroupConfigManager.class),
+ () -> 5_000_000
+ );
+
+ assertEquals(5_000_000, cache.dlqTopicMaxMessageBytes("test-topic"));
+ }
+
// Tests for topicName
@Test
@@ -714,7 +821,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertEquals(Optional.of("some-topic"), cache.topicName(topicId));
@@ -731,7 +839,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertEquals(Optional.empty(), cache.topicName(topicId));
@@ -748,7 +857,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertEquals(Optional.empty(), cache.topicName(topicId));
@@ -776,7 +886,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
ShareGroupDLQMetadataCacheHelper.TopicPartitionData data =
cache.topicPartitionData("test-topic");
@@ -807,7 +918,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
ShareGroupDLQMetadataCacheHelper.TopicPartitionData data =
cache.topicPartitionData("test-topic");
@@ -831,7 +943,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
ShareGroupDLQMetadataCacheHelper.TopicPartitionData data =
cache.topicPartitionData("test-topic");
@@ -864,7 +977,8 @@ public class ShareCoordinatorMetadataCacheHelperImplTest {
mockMetadataCache,
sharePartitionKey -> 0,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
ShareGroupDLQMetadataCacheHelper.TopicPartitionData data =
cache.topicPartitionData("test-topic");
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQMetadataCacheHelper.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQMetadataCacheHelper.java
index bbc3949cbdc..b372dccb0ca 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQMetadataCacheHelper.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQMetadataCacheHelper.java
@@ -84,6 +84,16 @@ public interface ShareGroupDLQMetadataCacheHelper {
*/
boolean containsTopic(String topic);
+ /**
+ * Get the max message size, in bytes, allowed to be produced to the given
topic. Honors
+ * a topic-level {@code max.message.bytes} override when present,
otherwise falls back to
+ * the broker's configured {@code message.max.bytes}.
+ *
+ * @param topic The name of the topic
+ * @return The maximum message size, in bytes, allowed for the topic
+ */
+ int dlqTopicMaxMessageBytes(String topic);
+
/**
* Get all nodes in the kafka cluster encapsulated in the {@link Node}
object.
*
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 1a27dd7f0bc..8cad732fbad 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
@@ -35,6 +35,9 @@ import org.apache.kafka.common.message.ProduceRequestData;
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;
@@ -262,6 +265,17 @@ public class ShareGroupDLQStateManager {
// sender thread when the produce request is built. Memoized: set once
and reused for every (re)send,
// so retries never re-fetch.
private volatile Map<Long, Record> resolvedRecordData = Map.of();
+ // The next offset that has not yet been included in a produce
request. Starts at
+ // param.firstOffset() and advances past whatever topicProduceData()
managed to fit within
+ // dlqTopicMaxMessageBytes() on each successful send, so a range that
doesn't fit in a single
+ // batch is sent as a sequence of produce requests instead of failing.
+ private volatile long nextOffsetToSend;
+ // The last offset included in the most recently built produce request
(see topicProduceData()),
+ // read back in handleProduceResponse() to decide whether more offsets
remain to be sent.
+ private volatile long lastOffsetIncludedThisRound;
+ // The dlqTopicMaxMessageBytes() value used to build the most recent
topicProduceData(), cached
+ // 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";
@@ -279,6 +293,8 @@ public class ShareGroupDLQStateManager {
) {
this.param = param;
this.result = result;
+ this.nextOffsetToSend = param.firstOffset();
+ this.lastOffsetIncludedThisRound = param.firstOffset() - 1;
this.createTopicsBackoff = new ExponentialBackoffManager(
maxRPCRetryAttempts,
backoffMs,
@@ -333,6 +349,13 @@ public class ShareGroupDLQStateManager {
.setName(TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG)
.setValue("true");
topicConfigs.add(enableDLQConfig);
+ // Every DLQ record's timestamp is explicitly set to the DLQ write
time (see topicProduceData()),
+ // so pin CreateTime here rather than letting the topic inherit
the cluster's broker-level
+ // log.message.timestamp.type default, which - if LogAppendTime -
would silently overwrite it.
+ CreateTopicsRequestData.CreatableTopicConfig timestampTypeConfig =
new CreateTopicsRequestData.CreatableTopicConfig()
+ .setName(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG)
+ .setValue(TimestampType.CREATE_TIME.name);
+ topicConfigs.add(timestampTypeConfig);
CreateTopicsRequestData.CreatableTopicCollection topicCollection =
new CreateTopicsRequestData.CreatableTopicCollection();
topicCollection.add(new CreateTopicsRequestData.CreatableTopic()
@@ -379,29 +402,55 @@ public class ShareGroupDLQStateManager {
// 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();
+ // 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<>();
- for (long i = param.firstOffset(); i <= param.lastOffset(); i++) {
+ int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
+ Long baseTimestamp = null;
+ for (long offset = nextOffsetToSend; offset <= param.lastOffset();
offset++) {
long timestamp = time.hiResClockMs();
ByteBuffer key = null;
ByteBuffer value = null;
- Record record = originalRecordData.get(i);
+ Record record = originalRecordData.get(offset);
if (record != null) {
key = record.hasKey() ? record.key() : null;
value = record.hasValue() ? record.value() : null;
}
- simpleRecords.add(new SimpleRecord(timestamp, key, value,
headers(i)));
+ Header[] recordHeaders = headers(offset);
+ 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;
+ }
}
+ 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.
+ this.lastMaxMessageBytes = maxMessageBytes;
MemoryRecords records = MemoryRecords.withRecords(
Compression.NONE,
simpleRecords.toArray(new SimpleRecord[]{})
);
- // Update the metric to say a new request is created to se sent.
This might not be the
- // actual RPC count as we coalesce the requests before sending.
- shareGroupMetrics.recordDLQProduce(param.groupId());
-
return new ProduceRequestData.TopicProduceData()
.setName(dlqTopicPartitionData.topicName())
.setTopicId(dlqTopicPartitionData.topicId().get())
@@ -412,6 +461,18 @@ public class ShareGroupDLQStateManager {
));
}
+ int dlqTopicMaxMessageBytes() {
+ return
cacheHelper.dlqTopicMaxMessageBytes(dlqTopicPartitionData.topicName());
+ }
+
+ int lastMaxMessageBytes() {
+ return lastMaxMessageBytes;
+ }
+
+ void recordProduceMetric() {
+ shareGroupMetrics.recordDLQProduce(param.groupId());
+ }
+
public Node dlqPartitionLeaderNode() {
return this.dlqPartitionLeaderNode;
}
@@ -647,9 +708,16 @@ public class ShareGroupDLQStateManager {
switch (error) {
case NONE:
LOG.debug("Successfully produced records {} to dlq
topic node {}.", this, dlqPartitionLeaderNode());
-
shareGroupMetrics.recordDLQRecordWrite(param.groupId(), (int)
(param.lastOffset() - param.firstOffset() + 1));
+
shareGroupMetrics.recordDLQRecordWrite(param.groupId(), (int)
(lastOffsetIncludedThisRound - nextOffsetToSend + 1));
produceRequestBackoff.resetAttempts();
- this.result.complete(null);
+ if (lastOffsetIncludedThisRound <
param.lastOffset()) {
+ // Only part of the offset range fit within
dlqTopicMaxMessageBytes - continue
+ // sending the remainder as a follow-up
produce request instead of completing.
+ nextOffsetToSend = lastOffsetIncludedThisRound
+ 1;
+ addRequestToNodeMap(dlqPartitionLeaderNode(),
this);
+ } else {
+ this.result.complete(null);
+ }
break;
case NOT_LEADER_OR_FOLLOWER:
@@ -796,27 +864,32 @@ public class ShareGroupDLQStateManager {
// 10. at this point P2, P3, etc. could be sent as a combined
request thus achieving batching.
final Set<Node> sending = new HashSet<>();
final Set<Node> emptyNodes = new HashSet<>(); // Nodes for which
no coalesced handler was found.
+ // Handlers that didn't fit within this round's
dlqTopicMaxMessageBytes budget for their
+ // destination partition - re-added to the node map below (after
nodeRPCMap.remove(node),
+ // which would otherwise wipe them out) so they're retried on a
subsequent tick.
+ final List<ProduceRequestHandler> deferredForNextRound = new
ArrayList<>();
synchronized (nodeMapLock) {
nodeRPCMap.forEach((destNode, handlers) -> {
// this condition causes requests of same type and same
destination node
// to not be sent immediately but get batched
if (!inFlight.contains(destNode)) {
CoalesceResults results =
coalesceProduceRequests(handlers);
- if (results.liveHandlers.isEmpty()) {
+
deferredForNextRound.addAll(results.deferredHandlers());
+ if (results.liveHandlers().isEmpty()) {
emptyNodes.add(destNode);
return;
}
requests.add(new RequestAndCompletionHandler(
time.milliseconds(),
destNode,
- results.request,
+ results.request(),
response -> {
inFlight.remove(destNode);
// now the combined request has completed
// we need to create responses for individual
// requests which composed the combined request
- results.liveHandlers.forEach(handler ->
handler.onComplete(response));
+ results.liveHandlers().forEach(handler ->
handler.onComplete(response));
wakeup();
}));
sending.add(destNode);
@@ -833,6 +906,8 @@ public class ShareGroupDLQStateManager {
});
} // close of synchronized context
+ deferredForNextRound.forEach(handler ->
addRequestToNodeMap(handler.dlqPartitionLeaderNode(), handler));
+
return requests;
}
@@ -873,7 +948,8 @@ public class ShareGroupDLQStateManager {
// Visibility for tests
record CoalesceResults(
AbstractRequest.Builder<? extends AbstractRequest> request,
- List<ProduceRequestHandler> liveHandlers
+ List<ProduceRequestHandler> liveHandlers,
+ List<ProduceRequestHandler> deferredHandlers
) {
}
@@ -887,18 +963,46 @@ public class ShareGroupDLQStateManager {
// that target the same (topic, partition) - and then build a single
produce request from that map.
Map<Uuid, String> topicNames = new HashMap<>();
Map<Uuid, Map<Integer, List<MemoryRecords>>>
recordsByTopicAndPartition = new LinkedHashMap<>();
+ // Running merged-batch size per (topic, partition), used to defer
handlers that would push a
+ // merged batch over dlqTopicMaxMessageBytes rather than sending an
oversized request. Each
+ // handler's own batch is already <= the limit (see
topicProduceData()), so this only ever needs
+ // to decide how many whole handlers fit - the first handler for a
given partition is always
+ // admitted, guaranteeing progress even if it alone is at the limit.
+ Map<Uuid, Map<Integer, Integer>> runningSizeByTopicAndPartition = new
HashMap<>();
List<ProduceRequestHandler> liveHandlers = new
ArrayList<>(handlers.size());
+ List<ProduceRequestHandler> deferredHandlers = new ArrayList<>();
handlers.forEach(handler -> {
try {
ProduceRequestData.TopicProduceData topicProduceData =
handler.topicProduceData();
Uuid topicId = topicProduceData.topicId();
+ int maxMessageBytes = handler.lastMaxMessageBytes();
+ Map<Integer, Integer> runningSizeByPartition =
+ runningSizeByTopicAndPartition.computeIfAbsent(topicId, k
-> new HashMap<>());
+
+ boolean fits =
topicProduceData.partitionData().stream().allMatch(partitionData -> {
+ int runningSize =
runningSizeByPartition.getOrDefault(partitionData.index(), 0);
+ int recordsSize = partitionData.records().sizeInBytes();
+ return runningSize == 0 || runningSize + recordsSize <=
maxMessageBytes;
+ });
+
+ if (!fits) {
+ deferredHandlers.add(handler);
+ return;
+ }
+
topicNames.putIfAbsent(topicId, topicProduceData.name());
Map<Integer, List<MemoryRecords>> partitionRecords =
recordsByTopicAndPartition.computeIfAbsent(topicId, k ->
new LinkedHashMap<>());
- topicProduceData.partitionData().forEach(partitionData ->
- partitionRecords.computeIfAbsent(partitionData.index(), k
-> new ArrayList<>())
- .add((MemoryRecords) partitionData.records()));
+ topicProduceData.partitionData().forEach(partitionData -> {
+ MemoryRecords records = (MemoryRecords)
partitionData.records();
+ partitionRecords.computeIfAbsent(partitionData.index(), k
-> new ArrayList<>()).add(records);
+ runningSizeByPartition.merge(partitionData.index(),
records.sizeInBytes(), Integer::sum);
+ });
liveHandlers.add(handler);
+ // Only counted once the handler's data is actually admitted
into the outgoing request -
+ // a handler that gets deferred here (see above) hasn't
produced anything yet, so it
+ // must not be counted, however many times it gets
re-evaluated across ticks.
+ handler.recordProduceMetric();
} catch (Exception exception) {
log.error("Unable to coalesce ProduceRequestData for handler
{}. It will be skipped from DLQ.", handler, exception);
handler.requestErrorResponse(exception);
@@ -925,7 +1029,8 @@ public class ShareGroupDLQStateManager {
return new CoalesceResults(
new ProduceRequest.Builder(ApiKeys.PRODUCE.latestVersion(),
ApiKeys.PRODUCE.latestVersion(), data),
- liveHandlers
+ liveHandlers,
+ deferredHandlers
);
}
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 a1b96b88660..1c37610f732 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
@@ -24,7 +24,9 @@ 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.message.CreateTopicsRequestData;
import org.apache.kafka.common.message.CreateTopicsResponseData;
import org.apache.kafka.common.message.ProduceRequestData;
import org.apache.kafka.common.message.ProduceResponseData;
@@ -113,6 +115,7 @@ class ShareGroupDLQStateManagerTest {
private static final Uuid SOURCE_TOPIC_ID = Uuid.randomUuid();
private static final Node DEFAULT_LEADER = new Node(0, HOST, PORT);
private static final LogReader MOCK_LOG_READER = mock(LogReader.class);
+ private static final int DEFAULT_MAX_MESSAGE_BYTES = 1024 * 1024;
private final MockTimer mockTimer = new MockTimer(MOCK_TIME);
private final ShareGroupMetrics mockMetrics =
mock(ShareGroupMetrics.class);
@@ -214,6 +217,7 @@ class ShareGroupDLQStateManagerTest {
List.of(leader)
));
when(helper.isShareGroupDlqCopyRecordEnabled(GROUP_ID)).thenReturn(false);
+
when(helper.dlqTopicMaxMessageBytes(anyString())).thenReturn(DEFAULT_MAX_MESSAGE_BYTES);
return helper;
}
@@ -560,6 +564,7 @@ class ShareGroupDLQStateManagerTest {
List.of(DEFAULT_LEADER)
));
when(cacheHelper.getClusterNodes()).thenReturn(List.of(DEFAULT_LEADER));
+
when(cacheHelper.dlqTopicMaxMessageBytes(anyString())).thenReturn(DEFAULT_MAX_MESSAGE_BYTES);
MockClient client = new MockClient(MOCK_TIME);
List<ProduceRequest> capturedProduces = new ArrayList<>();
@@ -613,6 +618,7 @@ class ShareGroupDLQStateManagerTest {
List.of(DEFAULT_LEADER)
));
when(cacheHelper.containsTopic(DLQ_TOPIC)).thenReturn(false);
+
when(cacheHelper.dlqTopicMaxMessageBytes(anyString())).thenReturn(DEFAULT_MAX_MESSAGE_BYTES);
MockClient client = new MockClient(MOCK_TIME);
List<ProduceRequest> capturedProduces = new ArrayList<>();
@@ -655,6 +661,26 @@ class ShareGroupDLQStateManagerTest {
verify(mockMetrics, never()).recordDLQProduceFailed(any());
}
+ @Test
+ public void testCreateTopicBuilderPinsCreateTimeAlongsideDlqEnableConfig()
throws Exception {
+ // DLQ record timestamps are explicitly set to the write time (see
topicProduceData()), so an
+ // auto-created DLQ topic must pin CreateTime rather than inheriting
the cluster's broker-level
+ // log.message.timestamp.type default, which - if LogAppendTime -
would silently discard it.
+ stateManager = builder().build();
+ ShareGroupDLQStateManager.ProduceRequestHandler handler =
newHandlerForCoalesceTest(stateManager, GROUP_ID, 0);
+
+ CreateTopicsRequest request = ((CreateTopicsRequest.Builder)
handler.createTopicBuilder()).build();
+ CreateTopicsRequestData.CreatableTopic topic =
request.data().topics().iterator().next();
+ assertEquals(DLQ_TOPIC, topic.name());
+
+ Map<String, String> configs = new HashMap<>();
+ topic.configs().forEach(c -> configs.put(c.name(), c.value()));
+ assertEquals(Map.of(
+ TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG, "true",
+ TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "CreateTime"
+ ), configs);
+ }
+
@Test
public void testDlqCreateTopicFatalErrorFailsFuture() throws Exception {
ShareGroupDLQMetadataCacheHelper cacheHelper =
mock(ShareGroupDLQMetadataCacheHelper.class);
@@ -760,6 +786,7 @@ class ShareGroupDLQStateManagerTest {
Optional.of(DLQ_TOPIC_ID),
List.of(DEFAULT_LEADER)
));
+
when(cacheHelper.dlqTopicMaxMessageBytes(anyString())).thenReturn(DEFAULT_MAX_MESSAGE_BYTES);
Timer realTimer = new SystemTimerReaper("shareGroupDLQTestTimer",
new SystemTimer("shareGroupDLQTestTimer"));
@@ -866,6 +893,52 @@ class ShareGroupDLQStateManagerTest {
}
}
+ @Test
+ public void
testDlqChunksOverMaxMessageBytesAcrossMultipleProduceRequests() throws
Exception {
+ // param() spans 3 offsets (0-2). With a 1-byte budget,
topicProduceData() still always
+ // includes at least one record per request (see topicProduceData()'s
single-record floor),
+ // so the range must be sent as three sequential produce requests
instead of failing or
+ // sending an oversized one.
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+ when(cacheHelper.dlqTopicMaxMessageBytes(anyString())).thenReturn(1);
+
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ MockClient.RequestMatcher captureProduce = body -> {
+ if (body instanceof ProduceRequest pr) {
+ capturedProduces.add(pr);
+ return true;
+ }
+ return false;
+ };
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+
+ stateManager =
builder().withClient(client).withCacheHelper(cacheHelper).build();
+ stateManager.start();
+ assertNull(stateManager.dlq(param()).get(10, TimeUnit.SECONDS));
+
+ assertEquals(3, capturedProduces.size(), "The 3-offset range must be
split into 3 chunked requests");
+ Map<String, String> sharedHeaders = Map.of(
+ HEADER_DLQ_ERRORS_TOPIC, "source-topic",
+ HEADER_DLQ_ERRORS_PARTITION, "0",
+ HEADER_DLQ_ERRORS_GROUP, GROUP_ID,
+ HEADER_DLQ_ERRORS_DELIVERY_COUNT, "1",
+ HEADER_DLQ_ERRORS_MESSAGE, "simulated cause"
+ );
+ for (int offset = 0; offset < 3; offset++) {
+ assertDlqProduceRecordHeaders(capturedProduces.get(offset), Map.of(
+ 0, new ExpectedDlqPartition(offset, offset, sharedHeaders,
List.of(), List.of())
+ ));
+ }
+
+ // Each chunk goes through topicProduceData()/generateRequests()
independently.
+ verify(mockMetrics, times(3)).recordDLQProduce(GROUP_ID);
+ verify(mockMetrics, times(3)).recordDLQRecordWrite(GROUP_ID, 1);
+ verify(mockMetrics, never()).recordDLQProduceFailed(any());
+ }
+
@Test
public void testDlqProduceFatalErrorFailsFuture() throws Exception {
MockClient client = new MockClient(MOCK_TIME);
@@ -1571,6 +1644,59 @@ class ShareGroupDLQStateManagerTest {
assertEquals(DLQ_TOPIC_ID,
request.data().topicData().iterator().next().topicId());
}
+ @Test
+ public void
testCoalesceProduceRequestsDefersHandlersThatExceedPartitionBudget() throws
Exception {
+ // Three single-offset handlers targeting the same DLQ partition
(default cache helper: single
+ // DLQ partition). With a 1-byte budget, the first handler for the
partition is still admitted
+ // (topicProduceData() always includes at least one record), but
merging a second or third
+ // handler's records in would exceed the budget - they must be
deferred, not failed.
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+ when(cacheHelper.dlqTopicMaxMessageBytes(anyString())).thenReturn(1);
+ stateManager = builder().withCacheHelper(cacheHelper).build();
+
+ ShareGroupDLQStateManager.ProduceRequestHandler h0 =
newHandlerForCoalesceTest(stateManager, GROUP_ID, 0);
+ ShareGroupDLQStateManager.ProduceRequestHandler h1 =
newHandlerForCoalesceTest(stateManager, GROUP_ID, 1);
+ ShareGroupDLQStateManager.ProduceRequestHandler h2 =
newHandlerForCoalesceTest(stateManager, GROUP_ID, 2);
+ h0.populateDLQTopicData();
+ h1.populateDLQTopicData();
+ h2.populateDLQTopicData();
+
+ ShareGroupDLQStateManager.CoalesceResults round1 =
+ ShareGroupDLQStateManager.coalesceProduceRequests(List.of(h0, h1,
h2));
+
+ assertEquals(List.of(h0), round1.liveHandlers(),
+ "Only the first handler for the partition should be admitted
within the byte budget");
+ assertEquals(List.of(h1, h2), round1.deferredHandlers(),
+ "Handlers that don't fit within the budget must be deferred, not
failed");
+ // Deferred handlers haven't produced anything yet, so the metric must
count only h0.
+ verify(mockMetrics, times(1)).recordDLQProduce(GROUP_ID);
+
+ // The resulting request must still be well-formed, containing only
the admitted handler's record.
+ ProduceRequest request = ((ProduceRequest.Builder)
round1.request()).build();
+ ProduceRequestData.TopicProduceData topic =
request.data().topicData().iterator().next();
+ assertEquals(1, topic.partitionData().size());
+ assertEquals(1, recordCount((MemoryRecords)
topic.partitionData().get(0).records()));
+
+ // Simulate the next tick: h1 and h2 were re-added to the node map
(real generateRequests()
+ // re-enqueues deferredHandlers via addRequestToNodeMap) and get
coalesced again on their own.
+ // h1 now gets the free first-handler pass; h2 is deferred a second
time.
+ ShareGroupDLQStateManager.CoalesceResults round2 =
+ ShareGroupDLQStateManager.coalesceProduceRequests(List.of(h1, h2));
+ assertEquals(List.of(h1), round2.liveHandlers());
+ assertEquals(List.of(h2), round2.deferredHandlers());
+ // h2 has now been deferred twice (round1 and round2) without ever
being admitted - its
+ // repeated re-evaluation must not inflate the metric on its own.
+ verify(mockMetrics, times(2)).recordDLQProduce(GROUP_ID);
+
+ // Round 3: h2 is alone and finally gets admitted.
+ ShareGroupDLQStateManager.CoalesceResults round3 =
+ ShareGroupDLQStateManager.coalesceProduceRequests(List.of(h2));
+ assertEquals(List.of(h2), round3.liveHandlers());
+ assertEquals(List.of(), round3.deferredHandlers());
+ // Exactly one recordDLQProduce per handler, total - not once per
deferral re-evaluation.
+ verify(mockMetrics, times(3)).recordDLQProduce(GROUP_ID);
+ }
+
// --- DLQ record with copy record enabled ---
private static FetchDataInfo recordsInfo(SimpleRecord... records) {
