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