lianetm commented on code in PR #16885:
URL: https://github.com/apache/kafka/pull/16885#discussion_r1746966454
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java:
##########
@@ -182,6 +215,234 @@ public CompletableFuture<Map<TopicPartition,
OffsetAndTimestampInternal>> fetchO
result.fetchedOffsets));
}
+ /**
+ * Update fetch positions for assigned partitions that do not have a
position. This will:
+ * <ul>
+ * <li>check if all assigned partitions already have fetch positions
and return right away if that's the case</li>
+ * <li>trigger an async request to validate positions (detect log
truncation)</li>
+ * <li>fetch committed offsets if enabled, and use the response to
update the positions</li>
+ * <li>fetch partition offsets for partitions that may still require a
position, and use the response to
+ * update the positions</li>
+ * </ul>
+ *
+ * @param deadlineMs Time in milliseconds when the triggering application
event expires. Any error received after
+ * this will be saved, and used to complete the result
exceptionally on the next call to this
+ * function.
+ * @return Future that will complete with a boolean indicating if all
assigned partitions have positions (based
+ * on {@link SubscriptionState#hasAllFetchPositions()}). It will complete
immediately, with true, if all positions
+ * are already available. If some positions are missing, the future will
complete once the offsets are retrieved and positions are updated.
+ */
+ public CompletableFuture<Boolean> updateFetchPositions(long deadlineMs) {
+ CompletableFuture<Boolean> result = new CompletableFuture<>();
+
+ try {
+ if (maybeCompleteWithPreviousException(result)) {
+ return result;
+ }
+
+ validatePositionsIfNeeded();
+
+ if (subscriptionState.hasAllFetchPositions()) {
+ // All positions are already available
+ result.complete(true);
+ return result;
+ }
+
+ // Some positions are missing, so trigger requests to fetch
offsets and update them.
+ updatePositionsWithOffsets(deadlineMs).whenComplete((__, error) ->
{
+ if (error != null) {
+ result.completeExceptionally(error);
+ } else {
+ result.complete(subscriptionState.hasAllFetchPositions());
+ }
+ });
+
+ } catch (Exception e) {
+ result.completeExceptionally(maybeWrapAsKafkaException(e));
+ }
+ return result;
+ }
+
+ private boolean
maybeCompleteWithPreviousException(CompletableFuture<Boolean> result) {
+ Throwable cachedException =
cachedUpdatePositionsException.getAndSet(null);
+ if (cachedException != null) {
+ result.completeExceptionally(cachedException);
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Generate requests to fetch offsets, and update positions once a
response is received.
+ */
+ private CompletableFuture<Void> updatePositionsWithOffsets(long
deadlineMs) {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+
+ cacheExceptionIfEventExpired(result, deadlineMs);
+
+ // Reset positions using committed offsets retrieved from the group
coordinator, for any
+ // partitions which do not have a valid position and are not awaiting
reset. This will
+ // trigger an OffsetFetch request and update positions with the
offsets retrieved.
+ if (commitRequestManager != null) {
+ CompletableFuture<Void> initWithCommittedOffsetsResult =
initWithCommittedOffsetsIfNeeded(deadlineMs);
+ initWithCommittedOffsetsResult.whenComplete((__, error) -> {
+ if (error == null) {
+ initWithPartitionOffsetsIfNeeded(result);
+ } else {
+ result.completeExceptionally(error);
+ }
+ });
+ } else {
+ initWithPartitionOffsetsIfNeeded(result);
+ }
+ return result;
+ }
+
+ /**
+ * Save exception that may occur while updating fetch positions. Note that
since the update fetch positions
+ * is triggered asynchronously, errors may be found when the triggering
UpdateFetchPositionsEvent has already
+ * expired. In that case, the exception is saved in memory, to be thrown
when processing the following
+ * UpdateFetchPositionsEvent.
+ *
+ * @param result Update fetch positions future to get the exception
from (if any)
+ * @param deadlineMs Deadline of the triggering application event, used to
identify if the event has already
+ * expired when the error in the result future occurs.
+ */
+ private void cacheExceptionIfEventExpired(CompletableFuture<Void> result,
long deadlineMs) {
+ result.whenComplete((__, error) -> {
+ boolean updatePositionsExpired = time.milliseconds() >= deadlineMs;
+ if (error != null && updatePositionsExpired) {
+ cachedUpdatePositionsException.set(error);
+ }
+ });
+ }
+
+ /**
+ * If there are partitions still needing a position and a reset policy is
defined, request
+ * reset using the default policy.
+ *
+ * @param result Future that will complete when the reset operation
completes.
+ * @throws NoOffsetForPartitionException If no reset strategy is configured
+ */
+ private void initWithPartitionOffsetsIfNeeded(CompletableFuture<Void>
result) {
+ try {
+ // Mark partitions that need reset, using the configured reset
strategy. If no
+ // strategy is defined, this will raise a
NoOffsetForPartitionException exception.
+ subscriptionState.resetInitializingPositions();
+ } catch (Exception e) {
+ result.completeExceptionally(e);
+ return;
+ }
+
+ // For partitions awaiting reset, generate a ListOffset request to
retrieve the partition
+ // offsets according to the strategy (ex. earliest, latest), and
update the positions.
+ resetPositionsIfNeeded().whenComplete((resetResult, error) -> {
+ if (error == null) {
+ result.complete(null);
+ } else {
+ result.completeExceptionally(error);
+ }
+ });
+ }
+
+ /**
+ * Fetch the committed offsets for partitions that require initialization.
Use them to set
+ * the fetch positions in the subscription state.
+ *
+ * @throws TimeoutException If offsets could not be retrieved within the
timeout
+ */
+ private CompletableFuture<Void> initWithCommittedOffsetsIfNeeded(long
deadlineMs) {
+ final Set<TopicPartition> initializingPartitions =
subscriptionState.initializingPartitions();
+
+ if (initializingPartitions.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ log.debug("Refreshing committed offsets for partitions {}",
initializingPartitions);
+ CompletableFuture<Void> result = new CompletableFuture<>();
+
+ // The shorter the timeout provided to poll(), the more likely the
offsets fetch will time out. To handle
+ // this case, on the first attempt to fetch the committed offsets, a
FetchCommittedOffsetsEvent is created
+ // (with potentially a longer timeout) and stored. The event is used
for the first attempt, but in the
+ // case it times out, subsequent attempts will also use the event in
order to wait for the results.
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>
refreshWithCommitted;
+ if (!canReusePendingOffsetFetchEvent(initializingPartitions)) {
+ // Generate a new OffsetFetch request and update positions when a
response is received
+ final long fetchCommittedDeadlineMs = Math.max(deadlineMs,
time.milliseconds() + defaultApiTimeoutMs);
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>
fetchOffsets =
+ commitRequestManager.fetchOffsets(initializingPartitions,
fetchCommittedDeadlineMs);
+ refreshWithCommitted =
refreshOffsetsAndCompleteResultOnResponseReceived(fetchOffsets, result);
+ pendingOffsetFetchEvent = new
PendingFetchCommittedRequest(initializingPartitions, refreshWithCommitted);
+ } else {
+ // Reuse pending OffsetFetch request
+ refreshWithCommitted = pendingOffsetFetchEvent.result;
+ refreshWithCommitted.whenComplete((__, error) -> {
+ if (error == null) {
+ result.complete(null);
+ } else {
+ result.completeExceptionally(error);
+ }
+ });
+ }
+
+ return result;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>
refreshOffsetsAndCompleteResultOnResponseReceived(
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>
fetchCommittedFuture,
+ final CompletableFuture<Void> result) {
+ return fetchCommittedFuture.whenComplete((offsets, error) -> {
+ pendingOffsetFetchEvent = null;
Review Comment:
Yes, this changed shape with the latest changes, so I believe we don't have
such situation anymore. Now the pending goes to null
[here](https://github.com/apache/kafka/blob/2b5de7dbc5a5ac9a8dad1bf99254dce8b09e11d0/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java#L372)
(even if the OffsetFetch future is somehow done already).
--
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]
