From eafae972f1ec304abb94156c3ac259fe171fab70 Mon Sep 17 00:00:00 2001 From: Sushant Mahajan Date: Fri, 10 Jul 2026 13:15:46 +0530 Subject: [PATCH 1/2] KAFKA-20771: Enforce max msg bytes in share group DLQ produce path. --- ...areCoordinatorMetadataCacheHelperImpl.java | 22 ++- .../scala/kafka/server/BrokerServer.scala | 4 +- ...oordinatorMetadataCacheHelperImplTest.java | 180 ++++++++++++++---- .../dlq/ShareGroupDLQMetadataCacheHelper.java | 10 + .../share/dlq/ShareGroupDLQStateManager.java | 139 ++++++++++++-- .../dlq/ShareGroupDLQStateManagerTest.java | 126 ++++++++++++ 6 files changed, 428 insertions(+), 53 deletions(-) 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) { From a82f415dc84d702c8852d7b7ff95ef0fb7a1fa4f Mon Sep 17 00:00:00 2001 From: Sushant Mahajan Date: Fri, 10 Jul 2026 16:27:53 +0530 Subject: [PATCH 2/2] add integ test to verify chunking --- .../consumer/ShareConsumerDLQTest.java | 137 ++++++++++++++++++ 1 file changed, 137 insertions(+) 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;