You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2026/07/10 07:48:26 UTC
[PR] KAFKA-20771: Enforce max msg bytes in share group DLQ produce path. [kafka]
smjn opened a new pull request, #22802:
URL: https://github.com/apache/kafka/pull/22802
- Add `dlqTopicMaxMessageBytes(topic`) to `ShareGroupDLQMetadataCacheHelper`,
honoring a topic-level max.message.bytes override or else falling back
to the broker's dynamically-reconfigurable message.max.bytes (via
`IntSupplier`, not a captured value).
- `ShareGroupDLQStateManager` now chunks
an oversized DLQ offset range across multiple produce requests with a
resumable per-handler integer, and defers (rather than fails) coalesced
handlers that would push a merged batch over the limit, re-enqueuing
them for a later round instead of sending oversized requests.
- Also, switches the per-record size check to an incremental calculation
(`DefaultRecord.sizeInBytes`) instead of re-serializing the batch-so-far
on every offset.
- Pins `message.timestamp.type=CreateTime` on
auto-created DLQ topics so timestamps aren't silently overwritten by
the cluster's broker-level default.
- Add new tests and update existing wherever applicable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20771: Enforce max msg bytes in share group DLQ produce path. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22802:
URL: https://github.com/apache/kafka/pull/22802#discussion_r3567260199
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -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<Long, Record> originalRecordData = resolvedRecordData;
+ int maxMessageBytes = dlqTopicMaxMessageBytes();
+ // In most cases the offset range is a single offset (see DLQ_MAX_FETCH_BYTES). Track the
+ // running batch size incrementally via DefaultRecord.sizeInBytes() - the same formula
+ // MemoryRecords itself uses - instead of re-serializing the whole batch-so-far on every
+ // offset, which would make this loop quadratic in the number of offsets.
List<SimpleRecord> simpleRecords = new ArrayList<>();
- for (long i = param.firstOffset(); i <= param.lastOffset(); i++) {
+ int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
+ Long baseTimestamp = null;
+ for (long offset = nextOffsetToSend; offset <= param.lastOffset(); offset++) {
long timestamp = time.hiResClockMs();
ByteBuffer key = null;
ByteBuffer value = null;
- Record record = originalRecordData.get(i);
+ Record record = originalRecordData.get(offset);
if (record != null) {
key = record.hasKey() ? record.key() : null;
value = record.hasValue() ? record.value() : null;
}
- simpleRecords.add(new SimpleRecord(timestamp, key, value, headers(i)));
+ Header[] recordHeaders = headers(offset);
+ if (baseTimestamp == null) {
+ baseTimestamp = timestamp;
+ }
+ int recordSize = DefaultRecord.sizeInBytes(simpleRecords.size(), timestamp - baseTimestamp, key, value, recordHeaders);
+
+ if (batchSize + recordSize > maxMessageBytes && !simpleRecords.isEmpty()) {
+ // Adding this record would exceed the limit and the batch already has at least one
+ // record - stop here and send the rest in a follow-up request.
+ break;
+ }
+ simpleRecords.add(new SimpleRecord(timestamp, key, value, recordHeaders));
+ batchSize += recordSize;
+ if (batchSize > maxMessageBytes) {
+ // A single record (with its DLQ headers) already exceeds the limit on its own;
+ // nothing to be gained by holding it back, so send it and let the broker
+ // enforce/report the ultimate limit for this one, rather than stalling forever.
+ break;
+ }
}
+ lastOffsetIncludedThisRound = nextOffsetToSend + simpleRecords.size() - 1;
Review Comment:
Will this work for compacted topics i.e. the offsets are missing?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20771: Enforce max msg bytes in share group DLQ produce path. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22802:
URL: https://github.com/apache/kafka/pull/22802#discussion_r3573540435
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -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<Long, Record> originalRecordData = resolvedRecordData;
+ int maxMessageBytes = dlqTopicMaxMessageBytes();
+ // In most cases the offset range is a single offset (see DLQ_MAX_FETCH_BYTES). Track the
+ // running batch size incrementally via DefaultRecord.sizeInBytes() - the same formula
+ // MemoryRecords itself uses - instead of re-serializing the whole batch-so-far on every
+ // offset, which would make this loop quadratic in the number of offsets.
List<SimpleRecord> simpleRecords = new ArrayList<>();
- for (long i = param.firstOffset(); i <= param.lastOffset(); i++) {
+ int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
+ Long baseTimestamp = null;
+ for (long offset = nextOffsetToSend; offset <= param.lastOffset(); offset++) {
long timestamp = time.hiResClockMs();
ByteBuffer key = null;
ByteBuffer value = null;
- Record record = originalRecordData.get(i);
+ Record record = originalRecordData.get(offset);
if (record != null) {
key = record.hasKey() ? record.key() : null;
value = record.hasValue() ? record.value() : null;
}
- simpleRecords.add(new SimpleRecord(timestamp, key, value, headers(i)));
+ Header[] recordHeaders = headers(offset);
+ if (baseTimestamp == null) {
+ baseTimestamp = timestamp;
+ }
+ int recordSize = DefaultRecord.sizeInBytes(simpleRecords.size(), timestamp - baseTimestamp, key, value, recordHeaders);
+
+ if (batchSize + recordSize > maxMessageBytes && !simpleRecords.isEmpty()) {
+ // Adding this record would exceed the limit and the batch already has at least one
+ // record - stop here and send the rest in a follow-up request.
+ break;
+ }
+ simpleRecords.add(new SimpleRecord(timestamp, key, value, recordHeaders));
+ batchSize += recordSize;
+ if (batchSize > maxMessageBytes) {
+ // A single record (with its DLQ headers) already exceeds the limit on its own;
+ // nothing to be gained by holding it back, so send it and let the broker
+ // enforce/report the ultimate limit for this one, rather than stalling forever.
+ break;
+ }
}
+ lastOffsetIncludedThisRound = nextOffsetToSend + simpleRecords.size() - 1;
+ // Cached so coalesceProduceRequests()'s partition-budget check can reuse the exact value
+ // used above to build this batch, instead of looking it up (and dynamically re-resolving
+ // it) a second time for the same round.
+ this.lastMaxMessageBytes = maxMessageBytes;
MemoryRecords records = MemoryRecords.withRecords(
Compression.NONE,
simpleRecords.toArray(new SimpleRecord[]{})
);
- // Update the metric to say a new request is created to se sent. This might not be the
- // actual RPC count as we coalesce the requests before sending.
- shareGroupMetrics.recordDLQProduce(param.groupId());
-
return new ProduceRequestData.TopicProduceData()
.setName(dlqTopicPartitionData.topicName())
.setTopicId(dlqTopicPartitionData.topicId().get())
Review Comment:
The `dlqTopicPartitionData` is a validated structure, so the `topicId` must be present. Should we use another structure to avoid using Optional? For example:
```java
record ValidatedTopicPartitionData(
String topicName,
int numPartitions,
Uuid topicId,
List<Node> partitionLeaderNodes
) { }
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20771: Enforce max msg bytes in share group DLQ produce path. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22802:
URL: https://github.com/apache/kafka/pull/22802#discussion_r3573338410
##########
core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java:
##########
@@ -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);
Review Comment:
Do we really need to initialize a `LogConfig`? It seems we could parse the `props` directly?
```java
try {
return Integer.parseInt(props.getProperty(TopicConfig.MAX_MESSAGE_BYTES_CONFIG));
} catch (Exception exe) {
return messageMaxBytesSupplier.getAsInt();
}
```
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -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<Long, Record> originalRecordData = resolvedRecordData;
+ int maxMessageBytes = dlqTopicMaxMessageBytes();
+ // In most cases the offset range is a single offset (see DLQ_MAX_FETCH_BYTES). Track the
+ // running batch size incrementally via DefaultRecord.sizeInBytes() - the same formula
+ // MemoryRecords itself uses - instead of re-serializing the whole batch-so-far on every
+ // offset, which would make this loop quadratic in the number of offsets.
List<SimpleRecord> simpleRecords = new ArrayList<>();
- for (long i = param.firstOffset(); i <= param.lastOffset(); i++) {
+ int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
+ Long baseTimestamp = null;
+ for (long offset = nextOffsetToSend; offset <= param.lastOffset(); offset++) {
long timestamp = time.hiResClockMs();
ByteBuffer key = null;
ByteBuffer value = null;
- Record record = originalRecordData.get(i);
+ Record record = originalRecordData.get(offset);
if (record != null) {
key = record.hasKey() ? record.key() : null;
value = record.hasValue() ? record.value() : null;
}
- simpleRecords.add(new SimpleRecord(timestamp, key, value, headers(i)));
+ Header[] recordHeaders = headers(offset);
+ if (baseTimestamp == null) {
+ baseTimestamp = timestamp;
+ }
+ int recordSize = DefaultRecord.sizeInBytes(simpleRecords.size(), timestamp - baseTimestamp, key, value, recordHeaders);
+
+ if (batchSize + recordSize > maxMessageBytes && !simpleRecords.isEmpty()) {
+ // Adding this record would exceed the limit and the batch already has at least one
+ // record - stop here and send the rest in a follow-up request.
+ break;
+ }
+ simpleRecords.add(new SimpleRecord(timestamp, key, value, recordHeaders));
+ batchSize += recordSize;
+ if (batchSize > maxMessageBytes) {
+ // A single record (with its DLQ headers) already exceeds the limit on its own;
+ // nothing to be gained by holding it back, so send it and let the broker
+ // enforce/report the ultimate limit for this one, rather than stalling forever.
+ break;
+ }
}
+ lastOffsetIncludedThisRound = nextOffsetToSend + simpleRecords.size() - 1;
Review Comment:
If an offset is missing due to log compaction, the current logic adds a record with a null key and null value to the DLQ. How would users distinguish between "this offset was compacted away" and "this offset actually failed and legitimately had a null key/value"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20771: Enforce max msg bytes in share group DLQ produce path. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22802:
URL: https://github.com/apache/kafka/pull/22802#discussion_r3569573568
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -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<Long, Record> originalRecordData = resolvedRecordData;
+ int maxMessageBytes = dlqTopicMaxMessageBytes();
+ // In most cases the offset range is a single offset (see DLQ_MAX_FETCH_BYTES). Track the
+ // running batch size incrementally via DefaultRecord.sizeInBytes() - the same formula
+ // MemoryRecords itself uses - instead of re-serializing the whole batch-so-far on every
+ // offset, which would make this loop quadratic in the number of offsets.
List<SimpleRecord> simpleRecords = new ArrayList<>();
- for (long i = param.firstOffset(); i <= param.lastOffset(); i++) {
+ int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
+ Long baseTimestamp = null;
+ for (long offset = nextOffsetToSend; offset <= param.lastOffset(); offset++) {
long timestamp = time.hiResClockMs();
ByteBuffer key = null;
ByteBuffer value = null;
- Record record = originalRecordData.get(i);
+ Record record = originalRecordData.get(offset);
if (record != null) {
key = record.hasKey() ? record.key() : null;
value = record.hasValue() ? record.value() : null;
}
- simpleRecords.add(new SimpleRecord(timestamp, key, value, headers(i)));
+ Header[] recordHeaders = headers(offset);
+ if (baseTimestamp == null) {
+ baseTimestamp = timestamp;
+ }
+ int recordSize = DefaultRecord.sizeInBytes(simpleRecords.size(), timestamp - baseTimestamp, key, value, recordHeaders);
+
+ if (batchSize + recordSize > maxMessageBytes && !simpleRecords.isEmpty()) {
+ // Adding this record would exceed the limit and the batch already has at least one
+ // record - stop here and send the rest in a follow-up request.
+ break;
+ }
+ simpleRecords.add(new SimpleRecord(timestamp, key, value, recordHeaders));
+ batchSize += recordSize;
+ if (batchSize > maxMessageBytes) {
+ // A single record (with its DLQ headers) already exceeds the limit on its own;
+ // nothing to be gained by holding it back, so send it and let the broker
+ // enforce/report the ultimate limit for this one, rather than stalling forever.
+ break;
+ }
}
+ lastOffsetIncludedThisRound = nextOffsetToSend + simpleRecords.size() - 1;
Review Comment:
Thanks @apoorvmittal10 for the insightful comment. The calculation above is being performed on simpleRecords which contains the records destined to the DLQ topic - which means it contains all records which have been dlq'ed (they may or may not contain key/value)
So this code works regardless.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20771: Enforce max msg bytes in share group DLQ produce path. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 merged PR #22802:
URL: https://github.com/apache/kafka/pull/22802
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org