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 1cbf5d16e02c5..2fe897fdd7180 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 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 void testManyRejectedRecordsAllWrittenToDlq() throws Exception {
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.
+ *
+ * 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.
+ *
+ *
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 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> dlqRecords = readDlqPartition(dlqTopic, 0, recordCount);
+ assertEquals(recordCount, dlqRecords.size(), "Unexpected number of records on the DLQ topic");
+ Set actualSourceOffsets = new HashSet<>();
+ for (ConsumerRecord 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 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> 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 6e0667cc6a92b..8d1fce822a221 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.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 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 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 boolean isDlqEnabledOnTopic(String topic) {
}
}
+ @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 = groupConfigManager.groupConfig(groupId);
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index d7502aaa039e3..f7f34c175370d 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 713a4d7890968..0ca8717a29302 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.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 void testConstructorThrowsErrorOnNullArgs() {
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 void testConstructorThrowsErrorOnNullArgs() {
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 void testConstructorThrowsErrorOnNullArgs() {
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 void testConstructorThrowsErrorOnNullArgs() {
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 void testContainsTopicReturnsFalseOnException() {
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 void testContainsTopicSuccess() {
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 void testShareCoordinatorReturnsNoNodeWhenNoInternalTopicInCache() {
mockMetadataCache,
func,
mock(ListenerName.class),
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertEquals(
@@ -167,7 +184,8 @@ public void testShareCoordinatorReturnsNoNodeIfTopicMetadataInvalid() {
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 void testShareCoordinatorReturnsNoNodeOnGetAliveNodeEmptyResponse() {
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 void testShareCoordinatorReturnsNoNodeOnGetAliveNodeException() {
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 void testShareCoordinatorSuccess() {
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 void testGetClusterNodesEmptyListOnException() {
mockMetadataCache,
func,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
when(mockMetadataCache.getAliveBrokerNodes(
@@ -455,7 +477,8 @@ public void testGetClusterNodesSuccess() {
mockMetadataCache,
func,
mockListenerName,
- mock(GroupConfigManager.class)
+ mock(GroupConfigManager.class),
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
List nodes = List.of(
@@ -488,7 +511,8 @@ public void testShareGroupDlqTopicReturnsEmptyWhenGroupConfigIsEmpty() {
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 void testShareGroupDlqTopicReturnsTopicName() {
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 void testIsShareGroupDlqCopyRecordEnabledReturnsFalseWhenGroupConfigEmpty
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 void testIsShareGroupDlqCopyRecordEnabledReturnsTrue() {
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 void testIsShareGroupDlqCopyRecordEnabledReturnsFalse() {
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 void testIsDlqAutoTopicCreateEnabledReturnsTrue() {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertTrue(cache.isDlqAutoTopicCreateEnabled());
@@ -594,7 +623,8 @@ public void testIsDlqAutoTopicCreateEnabledReturnsFalse() {
mock(MetadataCache.class),
sharePartitionKey -> 0,
mock(ListenerName.class),
- mockGroupConfigManager
+ mockGroupConfigManager,
+ () -> ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT
);
assertFalse(cache.isDlqAutoTopicCreateEnabled());
@@ -612,7 +642,8 @@ public void testShareGroupDlqTopicPrefixReturnsPrefix() {
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 void testShareGroupDlqTopicPrefixReturnsEmpty() {
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 void testIsDlqEnabledOnTopicReturnsFalseWhenTopicConfigNull() {
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 void testIsDlqEnabledOnTopicReturnsFalseWhenTopicConfigEmpty() {
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 void testIsDlqEnabledOnTopicReturnsTrue() {
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 void testIsDlqEnabledOnTopicReturnsFalse() {
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 void testTopicNameReturnsNameWhenPresent() {
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 void testTopicNameReturnsEmptyWhenNotPresent() {
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 void testTopicNameReturnsEmptyOnException() {
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 void testTopicPartitionDataReturnsFullData() {
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 void testTopicPartitionDataReturnsEmptyTopicIdWhenZeroUuid() {
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 void testTopicPartitionDataWithoutNumPartitions() {
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 void testTopicPartitionDataWithMissingPartitionLeader() {
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 bbc3949cbdc78..b372dccb0ca7d 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 record TopicPartitionData(
*/
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 1a27dd7f0bcd9..8cad732fbad89 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.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 @@ class ProduceRequestHandler implements RequestCompletionHandler {
// 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 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 ProduceRequestHandler(
) {
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 AbstractRequest.Builder createTopicBuilder() throws
.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 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 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 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 ProduceRequestData.TopicProduceData topicProduceData() {
));
}
+ 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 @@ private void handleProduceResponse(ClientResponse response) {
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 Collection generateRequests() {
// 10. at this point P2, P3, etc. could be sent as a combined request thus achieving batching.
final Set sending = new HashSet<>();
final Set 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 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 Collection generateRequests() {
});
} // close of synchronized context
+ deferredForNextRound.forEach(handler -> addRequestToNodeMap(handler.dlqPartitionLeaderNode(), handler));
+
return requests;
}
@@ -873,7 +948,8 @@ public void run() {
// Visibility for tests
record CoalesceResults(
AbstractRequest.Builder request,
- List liveHandlers
+ List liveHandlers,
+ List deferredHandlers
) {
}
@@ -887,18 +963,46 @@ static CoalesceResults coalesceProduceRequests(List handl
// that target the same (topic, partition) - and then build a single produce request from that map.
Map topicNames = new HashMap<>();
Map>> 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> runningSizeByTopicAndPartition = new HashMap<>();
List liveHandlers = new ArrayList<>(handlers.size());
+ List deferredHandlers = new ArrayList<>();
handlers.forEach(handler -> {
try {
ProduceRequestData.TopicProduceData topicProduceData = handler.topicProduceData();
Uuid topicId = topicProduceData.topicId();
+ int maxMessageBytes = handler.lastMaxMessageBytes();
+ Map 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> 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 @@ static CoalesceResults coalesceProduceRequests(List handl
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 a1b96b8866074..1c37610f73249 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.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 @@ private static ShareGroupDLQMetadataCacheHelper cacheHelper(Node leader) {
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 @@ public void testDlqTopicPrefixEmptyStringSkipsPrefixCheck() throws Exception {
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 capturedProduces = new ArrayList<>();
@@ -613,6 +618,7 @@ public void testDlqCreateTopicThenProduceSucceeds() throws Exception {
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 capturedProduces = new ArrayList<>();
@@ -655,6 +661,26 @@ public void testDlqCreateTopicThenProduceSucceeds() throws Exception {
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 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 @@ public void testDlqCreateTopicPartialFailuresThenSucceeds() throws Exception {
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 @@ public void testDlqProducePartialFailuresThenSucceeds() throws Exception {
}
}
+ @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 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 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 @@ public void testCoalesceProduceRequestsSkipsHandlerWhoseTopicProduceDataThrows()
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) {