lianetm commented on code in PR #14346:
URL: https://github.com/apache/kafka/pull/14346#discussion_r1319111762
##########
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:
For the record, even though the change for using committed offsets here is 2
lines of code replacing this TODO, it requires also a minor refactoring on
existing core components (ex. `ConsumerCoordinator`). So it will all follow
this, on a PR like "Adding support for using committed offsets when updating
fetch positions"
--
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
