lianetm commented on code in PR #14346:
URL: https://github.com/apache/kafka/pull/14346#discussion_r1319081696
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -209,6 +215,34 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
return ConsumerRecords.empty();
}
+ /**
+ * Set the fetch position to the committed position (if there is one) or
reset it using the
+ * offset reset policy the user has configured.
+ *
+ * @return true if the operation completed without timing out
+ * @throws org.apache.kafka.common.errors.AuthenticationException if
authentication fails. See the exception for more details
+ * @throws NoOffsetForPartitionException If no
offset is stored for a given partition and no offset reset policy is
+ * defined
+ */
+ private boolean updateFetchPositions() {
+ // If any partitions have been truncated due to a leader change, we
need to validate the offsets
+ ValidatePositionsApplicationEvent validatePositionsEvent = new
ValidatePositionsApplicationEvent();
+ eventHandler.add(validatePositionsEvent);
+
+ // TODO: integrate logic for refreshing committed offsets if available
Review Comment:
That's already implemented (in the integration branch), just included in a
different PR that should come after this, following the merge plan.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
