adikou commented on code in PR #22270:
URL: https://github.com/apache/kafka/pull/22270#discussion_r3305971392
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java:
##########
@@ -217,4 +289,60 @@ public interface ConsumerRebalanceListener {
default void onPartitionsLost(Collection<TopicPartition> partitions) {
onPartitionsRevoked(partitions);
}
+
+ /**
+ * Consumer-aware variant of {@link #onPartitionsLost(Collection)}.
+ *
+ * <p>This method receives a {@link RebalanceConsumer}, a restricted view
of the {@link Consumer}
+ * that exposes only operations safe during a rebalance. The {@code
RebalanceConsumer} is valid
+ * only for the duration of this callback; storing a reference and using
it later will throw
+ * {@link IllegalStateException}.
+ *
+ * <p>The default implementation delegates to {@link
#onPartitionsLost(Collection)}, which
+ * itself defaults to {@link #onPartitionsRevoked(Collection)}. The chain
always passes through
+ * the one-argument methods, not the two-argument variants. This preserves
the existing
+ * delegation behavior so that listeners overriding either one-argument
method see consistent
+ * results regardless of which entry point was used.
+ *
+ * <p>If you need custom behavior for both revocation and loss, override
both two-argument
+ * methods explicitly.
+ *
+ * <h4>Important difference from {@code onPartitionsRevoked}</h4>
+ *
+ * <p>Unlike {@link #onPartitionsRevoked(Collection, RebalanceConsumer)},
where the affected
+ * partitions are still part of the consumer's assignment until the
callback completes, the
+ * partitions passed to this method have <em>already been removed</em>
from the assignment
+ * before this callback fires. This means:
+ *
+ * <ul>
+ * <li>{@link RebalanceConsumer#commitSync() commitSync()}/{@link
RebalanceConsumer#commitAsync() commitAsync()}
+ * will not commit the current assignment, since the consumer no
longer includes the lost partitions.</li>
+ * <li>{@link RebalanceConsumer#commitSync(java.util.Map)
commitSync(offsets)}
+ * /{@link RebalanceConsumer#commitAsync(java.util.Map,
OffsetCommitCallback) commitAsync(offsets)}
+ * with the lost partitions explicitly will be rejected by the
broker since the consumer
+ * no longer owns them.</li>
+ * <li>{@link
RebalanceConsumer#position(org.apache.kafka.common.TopicPartition) position()}
and
+ * {@link
RebalanceConsumer#seek(org.apache.kafka.common.TopicPartition, long) seek()}
will
+ * fail for the lost partitions since they are not in the
assignment.</li>
+ * </ul>
+ *
+ * <p>The {@link RebalanceConsumer} does not add guardrails against these
operations on lost
+ * partitions. Users should be aware that offset management for lost
partitions is not
+ * possible in this callback.
+ *
+ * @param partitions The list of partitions that were assigned to the
consumer and now have been reassigned
+ * to other consumers. With both the Classic and
Consumer protocols, this will always include
+ * all partitions that were previously assigned to the
consumer.
+ * @param consumer A restricted view of the {@link Consumer} that is
only valid for the duration of this callback.
+ * See {@link RebalanceConsumer} for the set of
permitted operations.
+ * @throws org.apache.kafka.common.errors.WakeupException If raised
from a nested call to {@link KafkaConsumer}
Review Comment:
Fair point, switched in the other methods too
--
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]
