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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
