KAFKA-20771: Enforce max msg bytes in share group DLQ produce path.

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.
+     *
+     * <p>Phase 1: the source topic allows records up to {@code sourceMaxMessageBytes}, but the DLQ topic is
+     * configured with a third of that ({@code dlqMaxMessageBytes}). Three records, each sized so that any two
+     * of their (record-copy-enabled) DLQ copies together would exceed dlqMaxMessageBytes, are rejected together
+     * in a single commit - so SharePartition issues one DLQ call spanning the whole offset range (see
+     * ShareGroupDLQStateManagerTest for the underlying chunking logic). The DLQ produce path must then split
+     * that single call into multiple sequential produce requests (one record each) rather than failing or
+     * dropping any record; this confirms all 3 records still land on the DLQ topic, and that the DLQ
+     * produce-request count increased by at least 3 (one per chunk) - concrete proof splitting occurred, since
+     * without it the single oversized request would be rejected once (non-retriable) and no records would ever
+     * arrive.
+     *
+     * <p>Phase 2: the DLQ topic's {@code max.message.bytes} is then raised via {@code IncrementalAlterConfigs}
+     * to comfortably exceed what a fresh batch of 3 more (same-sized) records needs combined, and the reject
+     * scenario is repeated with that new batch. This confirms two more things: the broker picks up the raised
+     * limit dynamically (not a value cached at startup), and the chunking logic does not split unnecessarily
+     * once the budget is actually sufficient - the produce-request count must increase by exactly 1 for the
+     * second batch, not 3.
+     */
+    @ClusterTest
+    public void testDlqRespectsDlqTopicMaxMessageBytesNotEqToSourceTopic() throws Exception {
+        String groupId = "dlq-maxbytes-group";
+        String sourceTopic = "dlq-maxbytes-source";
+        String dlqTopic = "dlq.maxbytes";
+        int recordCount = 3;
+        int sourceMaxMessageBytes = 300_000;
+        int dlqMaxMessageBytes = sourceMaxMessageBytes / 3;
+        // Leave headroom below dlqMaxMessageBytes for the DLQ context headers/record-batch framing overhead,
+        // so a single record's DLQ copy cleanly fits under the limit but two together clearly don't.
+        int payloadSize = dlqMaxMessageBytes - 2_000;
+
+        try (Admin admin = createAdminClient()) {
+            admin.createTopics(Set.of(
+                new NewTopic(sourceTopic, 1, (short) 1)
+                    .configs(Map.of(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, Integer.toString(sourceMaxMessageBytes))),
+                new NewTopic(dlqTopic, 1, (short) 1)
+                    .configs(Map.of(
+                        TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG, "true",
+                        TopicConfig.MAX_MESSAGE_BYTES_CONFIG, Integer.toString(dlqMaxMessageBytes)))
+            )).all().get();
+        }
+
+        alterShareAutoOffsetReset(groupId, "earliest");
+        alterShareGroupConfig(groupId, GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+        // Record copy must be enabled - otherwise DLQ records carry headers only (tiny) and would never
+        // approach dlqMaxMessageBytes regardless of the source record size.
+        alterShareGroupConfig(groupId, GroupConfig.ERRORS_DEADLETTERQUEUE_COPY_RECORD_ENABLE_CONFIG, "true");
+
+        byte[] payload = new byte[payloadSize];
+        try (Producer<byte[], byte[]> producer = createProducer()) {
+            for (int i = 0; i < recordCount; i++) {
+                producer.send(new ProducerRecord<>(sourceTopic, 0, "key".getBytes(StandardCharsets.UTF_8), payload));
+            }
+            producer.flush();
+        }
+
+        // Reject all 3 records together in one commit, so SharePartition issues a single DLQ call spanning
+        // the whole offset range (a fresh, contiguous, single-fetch acquisition with no prior redeliveries
+        // maps to one cached in-flight batch, so one client-side reject commit produces one DLQ call).
+        rejectRecords(groupId, sourceTopic, recordCount);
+
+        // All 3 records must still reach the DLQ, split across multiple produce requests since no pairing
+        // of their DLQ copies fits within dlqMaxMessageBytes. Verified inline (rather than via
+        // verifyDlqTopicRecords()) since that helper hardcodes checking the copied value against the fixed
+        // "value" content produced by produceMessages()/produceTo(), not this test's large payload.
+        List<ConsumerRecord<byte[], byte[]>> dlqRecords = readDlqPartition(dlqTopic, 0, recordCount);
+        assertEquals(recordCount, dlqRecords.size(), "Unexpected number of records on the DLQ topic");
+        Set<Long> actualSourceOffsets = new HashSet<>();
+        for (ConsumerRecord<byte[], byte[]> record : dlqRecords) {
+            assertArrayEquals(payload, record.value(), "DLQ record value should be the copied payload");
+            assertEquals(groupId, headerValue(record, HEADER_DLQ_ERRORS_GROUP));
+            assertEquals(sourceTopic, headerValue(record, HEADER_DLQ_ERRORS_TOPIC));
+            assertEquals("0", headerValue(record, HEADER_DLQ_ERRORS_PARTITION));
+            actualSourceOffsets.add(Long.parseLong(Objects.requireNonNull(headerValue(record, HEADER_DLQ_ERRORS_OFFSET))));
+        }
+        assertEquals(expectedSourceOffsets(recordCount), actualSourceOffsets,
+            "DLQ records should cover every expected source offset");
+        verifyDlqMetrics(groupId, recordCount);
+
+        // Concrete proof that splitting - not some other mechanism - is why all 3 records arrived: there is
+        // exactly one logical DLQ call here (one contiguous reject batch, offsets 0-2), so any produce-request
+        // count above 1 for this group can only come from the resumable-cursor chunking logic splitting that
+        // one call into multiple sequential produce requests to stay within dlqMaxMessageBytes. Without it,
+        // the single oversized request would be rejected once (MESSAGE_TOO_LARGE is not retriable) and no
+        // records would ever reach the DLQ - contradicting the assertions above. >= rather than == tolerates
+        // an occasional extra retry (e.g. a transient network blip) without being flaky.
+        assertTrue(dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL, groupId) >= recordCount,
+            "Expected at least " + recordCount + " separate DLQ produce requests (one per chunk), was "
+                + dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL, groupId));
+        long produceCountBeforeRaise = dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL, groupId);
+
+        // Now raise the DLQ topic's max.message.bytes well above what all 3 (record-copy-enabled) DLQ copies
+        // need together, and repeat the same reject scenario with a fresh batch of 3 records. This confirms
+        // two things at once: dlqTopicMaxMessageBytes() picks up the change dynamically (it wraps a live
+        // topic-config lookup, not a value captured once at startup - see ShareCoordinatorMetadataCacheHelperImpl),
+        // and the chunking logic doesn't split unnecessarily when the budget is actually sufficient - the
+        // produce-request count must increase by exactly 1 (one request for the whole new batch), not 3.
+        int raisedDlqMaxMessageBytes = sourceMaxMessageBytes * 2;
+        ConfigResource dlqTopicResource = new ConfigResource(ConfigResource.Type.TOPIC, dlqTopic);
+        try (Admin admin = createAdminClient()) {
+            admin.incrementalAlterConfigs(
+                Map.of(dlqTopicResource, List.of(new AlterConfigOp(
+                    new ConfigEntry(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, Integer.toString(raisedDlqMaxMessageBytes)),
+                    AlterConfigOp.OpType.SET))),
+                new AlterConfigsOptions()
+            ).all().get();
+            waitForCondition(() -> {
+                Config config = admin.describeConfigs(List.of(dlqTopicResource)).all().get().get(dlqTopicResource);
+                ConfigEntry entry = config.get(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+                return entry != null && entry.value().equals(Integer.toString(raisedDlqMaxMessageBytes));
+            }, DEFAULT_MAX_WAIT_MS, 100L, () -> "Raised max.message.bytes did not propagate on the DLQ topic");
+        }
+
+        try (Producer<byte[], byte[]> producer = createProducer()) {
+            for (int i = 0; i < recordCount; i++) {
+                producer.send(new ProducerRecord<>(sourceTopic, 0, "key".getBytes(StandardCharsets.UTF_8), payload));
+            }
+            producer.flush();
+        }
+        rejectRecords(groupId, sourceTopic, recordCount);
+
+        List<ConsumerRecord<byte[], byte[]>> secondBatchDlqRecords = readDlqPartition(dlqTopic, 0, recordCount * 2);
+        assertEquals(recordCount * 2, secondBatchDlqRecords.size(),
+            "Unexpected number of records on the DLQ topic after the second batch");
+
+        waitForCondition(() -> dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL, groupId) == produceCountBeforeRaise + 1,
+            DEFAULT_MAX_WAIT_MS, 200L,
+            () -> "Expected exactly 1 additional DLQ produce request for the second batch (budget no longer forces "
+                + "chunking), count went from " + produceCountBeforeRaise + " to " + dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL, groupId));
+    }
+
     /**
      * Verifies the DLQ is NOT written when it is gated off, across the gating conditions:
      *   (a) no DLQ topic configured for the group;

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<SharePartitionKey, Integer> keyToPartitionMapper;
     private final ListenerName interBrokerListenerName;
     private final GroupConfigManager groupConfigManager;
+    private final IntSupplier messageMaxBytesSupplier;
     private final Logger log = LoggerFactory.getLogger(ShareCoordinatorMetadataCacheHelperImpl.class);
 
     public ShareCoordinatorMetadataCacheHelperImpl(
         MetadataCache metadataCache,
         Function<SharePartitionKey, Integer> keyToPartitionMapper,
         ListenerName interBrokerListenerName,
-        GroupConfigManager groupConfigManager
+        GroupConfigManager groupConfigManager,
+        IntSupplier messageMaxBytesSupplier
     ) {
         this.metadataCache = Objects.requireNonNull(metadataCache, "metadataCache must not be null");
         this.keyToPartitionMapper = Objects.requireNonNull(keyToPartitionMapper, "keyToPartitionMapper must not be null");
         this.interBrokerListenerName = Objects.requireNonNull(interBrokerListenerName, "interBrokerListenerName must not be null");
         this.groupConfigManager = Objects.requireNonNull(groupConfigManager, "groupConfigManager must not be null");
+        this.messageMaxBytesSupplier = Objects.requireNonNull(messageMaxBytesSupplier, "messageMaxBytesSupplier must not be null");
     }
 
     @Override
@@ -103,6 +107,22 @@ public 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> groupConfig = groupConfigManager.groupConfig(groupId);

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,