AndrewJSchofield commented on code in PR #23964:
URL: https://github.com/apache/kafka/pull/23964#discussion_r4392722635
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -704,6 +713,27 @@ private void handleProduceResponse(ClientResponse
response) {
}
}
+ /**
+ * Runs {@link #resolveRound()} on the fetch executor, so the log read
it starts never blocks the
+ * calling thread: round 1 is called from the thread completing a
share partition's persister
+ * write, and later rounds from the sender thread. Like {@link
#resolveRound()}, the returned
+ * future always completes normally, including when the executor
rejects the task, in which case
+ * this round is produced with headers only.
+ */
+ CompletableFuture<Void> resolveRoundAsync() {
+ CompletableFuture<Void> resolved = new CompletableFuture<>();
+ try {
+ fetchExecutor.execute(() ->
resolveRound().whenComplete((ignored, err) -> resolved.complete(null)));
+ } catch (RejectedExecutionException e) {
+ LOG.warn("Unable to schedule the fetch of original record data
for {}. " +
+ "DLQ records will be produced with headers only for this
round.", this, e);
+ this.resolvedRecordData = Map.of();
+ this.lastResolvedOffsetThisRound = param.lastOffset();
+ resolved.complete(null);
+ }
+ return resolved;
+ }
+
/**
* Resolves the source records for the CURRENT round only - the window
starting at
* {@link #nextOffsetToSend} - reading from the local log on the
calling thread and, for any
Review Comment:
"Calling thread" is incorrect now.
##########
core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java:
##########
@@ -111,17 +118,21 @@ 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)) {
+ // Only a topic-level override is read here; without one the broker's
configured
+ // message.max.bytes is used.
+ Object value = props == null ? null :
props.get(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+ if (value == null) {
return messageMaxBytesSupplier.getAsInt();
}
try {
- return new
LogConfig(props).getInt(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+ Integer maxMessageBytes = (Integer)
ConfigDef.parseType(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, value,
ConfigDef.Type.INT);
+ if (maxMessageBytes != null) {
Review Comment:
This can't be null. If it's bad, you're going to end up with an exception
instead.
##########
core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java:
##########
@@ -111,17 +118,21 @@ 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)) {
+ // Only a topic-level override is read here; without one the broker's
configured
+ // message.max.bytes is used.
+ Object value = props == null ? null :
props.get(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+ if (value == null) {
return messageMaxBytesSupplier.getAsInt();
}
try {
- return new
LogConfig(props).getInt(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+ Integer maxMessageBytes = (Integer)
ConfigDef.parseType(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, value,
ConfigDef.Type.INT);
Review Comment:
You will need to apply bounds checking to the returned value though I think.
##########
server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java:
##########
@@ -82,10 +110,9 @@ public CompletableFuture<Void>
enqueue(ShareGroupDLQRecordParameter param) {
@Override
public void stop() {
- try {
- stateManager.stop();
- } catch (Exception e) {
- log.error("Unable to stop DLQ state manager", e);
- }
+ // Drain the reads first - they enqueue onto the state manager when
they finish - and interrupt
+ // any which do not finish in time.
+ ThreadUtils.shutdownExecutorServiceQuietly(fetchExecutor,
FETCH_EXECUTOR_SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS);
Review Comment:
This just orphans enqueued but unstarted tasks without completing their
futures. You should capture the return value from `fetchExecutor.shutdownNow()`
and process the list of tasks to fail their futures.
--
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]
