apoorvmittal10 commented on code in PR #22601:
URL: https://github.com/apache/kafka/pull/22601#discussion_r3481773498
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -238,9 +250,18 @@ class ProduceRequestHandler implements
RequestCompletionHandler {
private static final Logger LOG =
LoggerFactory.getLogger(ShareGroupDLQStateManager.ProduceRequestHandler.class);
private final ExponentialBackoffManager createTopicsBackoff;
private final ExponentialBackoffManager produceRequestBackoff;
- private Node dlqPartitionLeaderNode;
- private int dlqDestinationPartition;
- private ShareGroupDLQMetadataCacheHelper.TopicPartitionData
dlqTopicPartitionData;
+ // These DLQ topic fields are written by populateDLQTopicData() and
read while building the
+ // produce request - both on the sender thread (from
dlqTopicExists()/handleCreateTopicsResponse()).
+ // Kept volatile defensively.
Review Comment:
I didn't understand why defensively, do we understand if there could be an
issue or not?
##########
server/src/main/java/org/apache/kafka/server/share/LogReader.java:
##########
@@ -42,4 +46,44 @@ LinkedHashMap<TopicIdPartition, LogReadResult> read(
Set<TopicIdPartition> partitionsToFetch,
LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes);
+
+ /**
+ * The outcome of an asynchronous read for a single partition.
Partial-data tolerant: {@code fetchDataInfo}
+ * holds whatever data could be read (its {@link FetchDataInfo#records}),
and {@code error} is
+ * {@link Errors#NONE} on success or the failure reason otherwise.
+ *
+ * @param fetchDataInfo The data read for the partition.
+ * @param error {@link Errors#NONE} on success, otherwise the read
failure.
+ */
+ record AsyncReadResult(FetchDataInfo fetchDataInfo, Errors error) {
+ }
+
+ /**
+ * Read records for the given partitions starting at the specified
offsets, combining the local read
+ * and - when {@code readRemote} is true and the requested data has been
tiered off the local log - the
+ * follow-up remote read into a single call.
+ *
+ * <p>This is the asynchronous, remote-aware counterpart to {@link #read}:
it returns one future per
+ * requested partition. Partitions whose data is available locally (or
whose local read failed) complete
+ * immediately; partitions whose data is in remote storage complete later,
once the remote read finishes
+ * on the remote storage reader pool, so the caller's thread is never
blocked on remote IO. When
+ * {@code readRemote} is false, tiered offsets are simply omitted from the
result rather than fetched.
+ *
+ * <p>Each per-partition result is partial-data tolerant (see {@link
AsyncReadResult}); the read never
+ * fails as a whole, allowing callers to use whatever records were
retrieved and skip the rest.
+ *
+ * @param fetchParams The fetch parameters (isolation
level, maxBytes, etc.)
+ * @param partitionsToFetch The set of partitions to fetch
+ * @param topicPartitionFetchOffsets The fetch offset per partition
+ * @param partitionMaxBytes The max bytes per partition
+ * @param readRemote Whether to follow tiered offsets to
the remote tier; when false,
+ * tiered offsets are skipped.
+ * @return A map from partition to a future of that partition's {@link
AsyncReadResult}.
+ */
+ Map<TopicIdPartition, CompletableFuture<AsyncReadResult>> readAsync(
Review Comment:
Will it make sense to return
`CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>` as response
so caller shouldn't worry about waiting for independent topic partition
completion futures? I think we should simplify.
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -238,9 +250,18 @@ class ProduceRequestHandler implements
RequestCompletionHandler {
private static final Logger LOG =
LoggerFactory.getLogger(ShareGroupDLQStateManager.ProduceRequestHandler.class);
private final ExponentialBackoffManager createTopicsBackoff;
private final ExponentialBackoffManager produceRequestBackoff;
- private Node dlqPartitionLeaderNode;
- private int dlqDestinationPartition;
- private ShareGroupDLQMetadataCacheHelper.TopicPartitionData
dlqTopicPartitionData;
+ // These DLQ topic fields are written by populateDLQTopicData() and
read while building the
+ // produce request - both on the sender thread (from
dlqTopicExists()/handleCreateTopicsResponse()).
+ // Kept volatile defensively.
+ private volatile Node dlqPartitionLeaderNode;
+ private volatile int dlqDestinationPartition;
+ private volatile ShareGroupDLQMetadataCacheHelper.TopicPartitionData
dlqTopicPartitionData;
+ // The original source records, resolved once before this handler is
enqueued (see resolveRecords()).
+ // Volatile because resolution runs off the sender thread - on the
calling thread for local offsets
+ // and, for tiered offsets, on the remote-storage reader pool - while
this value is read on the
+ // sender thread when the produce request is built. Memoized: set once
and reused for every (re)send,
+ // so retries never re-fetch.
+ private volatile Map<Long, Record> resolvedRecordData = Map.of();
Review Comment:
Why do we need instance variable as the response from resolveRecords() is
completableFuture which can send back the resolved records itself?
--
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]
