AndrewJSchofield commented on code in PR #22270:
URL: https://github.com/apache/kafka/pull/22270#discussion_r3305075817
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java:
##########
@@ -31,9 +31,23 @@
* of the members changes. This can occur when processes die, new process
instances are added or old instances come back to life after failure.
* Partition re-assignments can also be triggered by changes affecting the
subscribed topics (e.g. when the number of partitions is
* administratively adjusted).
+ *
+ * <h3>Consumer-Aware Callbacks</h3>
+ *
+ * Each callback method has two variants: a one-argument form that receives
only the affected partitions, and a
+ * two-argument form that additionally receives a {@link RebalanceConsumer} -
a restricted view of the
+ * {@link Consumer} exposing only operations that are safe during a rebalance.
The two-argument variants default
+ * to delegating to their one-argument counterparts, so existing
implementations continue to work unchanged.
* <p>
- * There are many uses for this functionality. One common use is saving
offsets in a custom store. By saving offsets in
- * the {@link #onPartitionsRevoked(Collection)} call we can ensure that any
time partition assignment changes
+ * When consumer access is needed during a rebalance (e.g. to commit offsets
or seek to external positions),
+ * prefer overriding the two-argument variants and using the provided {@link
RebalanceConsumer} instead of
+ * capturing the full {@code Consumer} reference externally. This avoids
accidental use of operations like
+ * {@code poll()}, {@code close()}, or {@code subscribe()} that could corrupt
consumer state mid-rebalance.
+ *
+ * <h3>Common Use Cases</h3>
+ *
+ * <p>There are many uses for this functionality. One common use is saving
offsets in a custom store. By saving offsets in
Review Comment:
nit: We tend not to use `<p>` just after the heading in the javadoc. Just
for consistency. See `KafkaConsumer` for example.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java:
##########
@@ -48,27 +62,29 @@
* whenever partition assignment changes.
* <p>
* Under normal conditions, if a partition is reassigned from one consumer to
another, then the old consumer will
- * always invoke {@link #onPartitionsRevoked(Collection) onPartitionsRevoked}
for that partition prior to the new consumer
- * invoking {@link #onPartitionsAssigned(Collection) onPartitionsAssigned} for
the same partition. So if offsets or other state is saved in the
- * {@link #onPartitionsRevoked(Collection) onPartitionsRevoked} call by one
consumer member, it will always be accessible by the time the
- * other consumer member taking over that partition and triggering its {@link
#onPartitionsAssigned(Collection) onPartitionsAssigned} callback to load the
state.
+ * always invoke {@link #onPartitionsRevoked(Collection, RebalanceConsumer)
onPartitionsRevoked} for that partition
+ * prior to the new consumer invoking {@link #onPartitionsAssigned(Collection,
RebalanceConsumer) onPartitionsAssigned}
+ * for the same partition. So if offsets or other state is saved durably in the
+ * {@link #onPartitionsRevoked(Collection, RebalanceConsumer)
onPartitionsRevoked} callback by one consumer member,
+ * it will always be accessible by the time the other consumer member taking
over that partition triggers its
+ * {@link #onPartitionsAssigned(Collection, RebalanceConsumer)
onPartitionsAssigned} callback to load the state.
* <p>
* You can think of revocation as a graceful way to give up ownership of a
partition. In some cases, the consumer may not have an opportunity to do so.
* For example, if the session times out, then the partitions may be
reassigned before we have a chance to revoke them gracefully.
- * For this case, we have a third callback {@link
#onPartitionsLost(Collection)}. The difference between this function and
- * {@link #onPartitionsRevoked(Collection)} is that upon invocation of {@link
#onPartitionsLost(Collection)}, the partitions
+ * For this case, we have a third callback {@link
#onPartitionsLost(Collection, RebalanceConsumer)}. The difference between this
function and
+ * {@link #onPartitionsRevoked(Collection, RebalanceConsumer)} is that upon
invocation of {@link #onPartitionsLost(Collection, RebalanceConsumer)}, the
partitions
* may already be owned by some other members in the group and therefore users
would not be able to commit its consumed offsets for example.
* Users could implement these two functions differently (by default,
- * {@link #onPartitionsLost(Collection)} will be calling {@link
#onPartitionsRevoked(Collection)} directly); for example, in the
- * {@link #onPartitionsLost(Collection)} we should not need to store the
offsets since we know these partitions are no longer owned by the consumer
+ * {@link #onPartitionsLost(Collection, RebalanceConsumer)} will be calling
{@link #onPartitionsRevoked(Collection, RebalanceConsumer)} directly); for
example, in the
+ * {@link #onPartitionsLost(Collection, RebalanceConsumer)} we should not need
to store the offsets since we know these partitions are no longer owned by the
consumer
* at that time.
* <p>
- * During a rebalance event, the {@link #onPartitionsAssigned(Collection)
onPartitionsAssigned} function will always be triggered exactly once when
- * the rebalance completes. That is, even if there is no newly assigned
partitions for a consumer member, its {@link #onPartitionsAssigned(Collection)
onPartitionsAssigned}
+ * During a rebalance event, the {@link #onPartitionsAssigned(Collection,
RebalanceConsumer) onPartitionsAssigned} function will always be triggered
exactly once when
+ * the rebalance completes. That is, even if there is no newly assigned
partitions for a consumer member, its {@link #onPartitionsAssigned(Collection,
RebalanceConsumer) onPartitionsAssigned}
Review Comment:
nit: Grammar. There ARE no newly assigned partitions.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java:
##########
@@ -89,31 +105,31 @@
* Here is pseudo-code for a callback implementation for saving offsets:
* <pre>
* {@code
- * public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {
- * private Consumer<?,?> consumer;
- *
- * public SaveOffsetsOnRebalance(Consumer<?,?> consumer) {
- * this.consumer = consumer;
- * }
- *
- * public void onPartitionsRevoked(Collection<TopicPartition>
partitions) {
+ * consumer.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
+ * @Override
+ * public void onPartitionsRevoked(Collection<TopicPartition>
partitions, RebalanceConsumer rebalanceConsumer) {
* // save the offsets in an external store using some custom code
not described here
* for(TopicPartition partition: partitions)
- * saveOffsetInExternalStore(consumer.position(partition));
+ *
saveOffsetInExternalStore(rebalanceConsumer.position(partition));
* }
*
+ * @Override
* public void onPartitionsLost(Collection<TopicPartition> partitions) {
Review Comment:
nit: In the example, I would override the two-argument variant.
--
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]
