AndrewJSchofield commented on code in PR #22270:
URL: https://github.com/apache/kafka/pull/22270#discussion_r3380345333
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java:
##########
@@ -89,33 +107,44 @@
* 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));
* }
*
- * public void onPartitionsLost(Collection<TopicPartition> partitions) {
+ * @Override
+ * public void onPartitionsLost(Collection<TopicPartition> partitions,
RebalanceConsumer rebalanceConsumer) {
* // do not need to save the offsets since these partitions are
probably owned by other consumers already
* }
*
- * public void onPartitionsAssigned(Collection<TopicPartition>
partitions) {
+ * @Override
+ * public void onPartitionsAssigned(Collection<TopicPartition>
partitions, RebalanceConsumer rebalanceConsumer) {
* // read the offsets from an external store using some custom code
not described here
* for(TopicPartition partition: partitions)
- * consumer.seek(partition,
readOffsetFromExternalStore(partition));
+ * rebalanceConsumer.seek(partition,
readOffsetFromExternalStore(partition));
* }
- * }
+ * });
+ * consumer.subscribe(List.of("topic-1", "topic-2"));
* }
* </pre>
+ *
+ * TODO(adikou): re-enable as a real {@code @deprecated} javadoc tag once
internal call sites have
+ * migrated off ConsumerRebalanceListener; left for now to avoid
-Werror.
+ * @deprecated Since 4.4, replaced by {@link RebalanceListener}. New code
should implement
+ * {@link RebalanceListener} directly and register it via
+ * {@code Consumer.setRebalanceListener(RebalanceListener)}. This
interface remains for
+ * source compatibility with pre-existing one-argument callback
implementations and will
+ * continue to be supported for the foreseeable future to give
users time to migrate.
+ *
+ * @see RebalanceListener
+ * @see RebalanceConsumer
*/
-public interface ConsumerRebalanceListener {
+// TODO(adikou): re-enable @Deprecated once internal call sites have migrated;
left commented for now to avoid -Werror.
Review Comment:
We do not tend to leave TODO in the code. I suggest creating a JIRA instead
for the follow-on work.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java:
##########
@@ -89,33 +107,44 @@
* 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));
* }
*
- * public void onPartitionsLost(Collection<TopicPartition> partitions) {
+ * @Override
+ * public void onPartitionsLost(Collection<TopicPartition> partitions,
RebalanceConsumer rebalanceConsumer) {
* // do not need to save the offsets since these partitions are
probably owned by other consumers already
* }
*
- * public void onPartitionsAssigned(Collection<TopicPartition>
partitions) {
+ * @Override
+ * public void onPartitionsAssigned(Collection<TopicPartition>
partitions, RebalanceConsumer rebalanceConsumer) {
* // read the offsets from an external store using some custom code
not described here
* for(TopicPartition partition: partitions)
- * consumer.seek(partition,
readOffsetFromExternalStore(partition));
+ * rebalanceConsumer.seek(partition,
readOffsetFromExternalStore(partition));
* }
- * }
+ * });
+ * consumer.subscribe(List.of("topic-1", "topic-2"));
* }
* </pre>
+ *
+ * TODO(adikou): re-enable as a real {@code @deprecated} javadoc tag once
internal call sites have
Review Comment:
Again, I wouldn't leave a TODO. We understand that `[1/N]` PRs are the first
of several pieces.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java:
##########
@@ -89,33 +107,44 @@
* 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() {
Review Comment:
This method is called `setRebalanceListener` in the KIP since the latest
revision.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/RebalanceListener.java:
##########
@@ -0,0 +1,288 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer;
+
+import org.apache.kafka.common.TopicPartition;
+
+import java.time.Duration;
+import java.util.Collection;
+
+/**
+ * A callback interface that the user can implement to trigger custom actions
when the set of partitions assigned to the
+ * consumer changes.
+ * <p>
+ * This is applicable when the consumer is having Kafka auto-manage group
membership. If the consumer directly assigns partitions,
+ * those partitions will never be reassigned and this callback is not
applicable.
+ * <p>
+ * When Kafka is managing the group membership, a partition re-assignment will
be triggered any time the members of the group change or the subscription
+ * 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 receives a {@link RebalanceConsumer}, a restricted
view of the {@link Consumer} that exposes
+ * only operations safe to invoke during a rebalance (e.g. offset commits,
seeks, position queries). The
+ * {@code RebalanceConsumer} is valid only for the duration of the callback;
storing a reference and using it later
+ * will throw {@link IllegalStateException}.
+ * <p>
+ * Prefer using the provided {@code RebalanceConsumer} instead of capturing
the full {@link 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>
+ *
+ * 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, RebalanceConsumer)} call we can
ensure that any time partition assignment changes
+ * the offset gets saved.
+ * <p>
+ * Another use is flushing out any kind of cache of intermediate results the
consumer may be keeping. For example,
+ * consider a case where the consumer is subscribed to a topic containing user
page views, and the goal is to count the
+ * number of page views per user for each five minute window. Let's say the
topic is partitioned by the user id so that
+ * all events for a particular user go to a single consumer instance. The
consumer can keep in memory a running
+ * tally of actions per user and only flush these out to a remote data store
when its cache gets too big. However if a
+ * partition is reassigned it may want to automatically trigger a flush of
this cache, before the new owner takes over
+ * consumption.
+ * <p>
+ * This callback will only execute in the user thread as part of the {@link
KafkaConsumer#poll(java.time.Duration) poll(Duration)} call
+ * 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, 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, 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, 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,
RebalanceConsumer) onPartitionsAssigned} function will always be triggered
exactly once when
+ * the rebalance completes. That is, even if there are no newly assigned
partitions for a consumer member, its {@link #onPartitionsAssigned(Collection,
RebalanceConsumer) onPartitionsAssigned}
+ * will still be triggered with an empty collection of partitions. As a result
this function can be used also to notify when a rebalance event has happened.
+ * With eager rebalancing, {@link #onPartitionsRevoked(Collection,
RebalanceConsumer)} will always be called at the start of a rebalance. On the
other hand, {@link #onPartitionsLost(Collection, RebalanceConsumer)}
+ * will only be called when there were non-empty partitions that were lost.
+ * With cooperative rebalancing, {@link #onPartitionsRevoked(Collection,
RebalanceConsumer)} and {@link #onPartitionsLost(Collection, RebalanceConsumer)}
+ * will only be triggered when there are non-empty partitions revoked or lost
from this consumer member during a rebalance event.
+ * <p>
+ * It is possible
+ * for a {@link org.apache.kafka.common.errors.WakeupException} or {@link
org.apache.kafka.common.errors.InterruptException}
+ * to be raised from one of these nested invocations. In this case, the
exception will be propagated to the current
+ * invocation of {@link KafkaConsumer#poll(java.time.Duration)} in which this
callback is being executed. This means it is not
+ * necessary to catch these exceptions and re-attempt to wakeup or interrupt
the consumer thread.
+ * Also if the callback function implementation itself throws an exception,
this exception will be propagated to the current
+ * invocation of {@link KafkaConsumer#poll(java.time.Duration)} as well.
+ * <p>
+ * Note that callbacks only serve as notification of an assignment change.
+ * They cannot be used to express acceptance of the change.
+ * Hence throwing an exception from a callback does not affect the assignment
in any way,
+ * as it will be propagated all the way up to the {@link
KafkaConsumer#poll(java.time.Duration)} call.
+ * If user captures the exception in the caller, the callback is still assumed
successful and no further retries will be attempted.
+ *
+ * <h3>Relation to {@link ConsumerRebalanceListener}</h3>
+ *
+ * {@code RebalanceListener} is the preferred entry point for new code. The
legacy {@link ConsumerRebalanceListener}
+ * interface extends this one and adds one-argument variants of each callback
for source compatibility with
+ * pre-existing implementations. New code should implement {@code
RebalanceListener} directly and register it via
+ * {@code Consumer.setRebalanceListener(RebalanceListener)}.
+ * <p>
+ *
+ * Here is pseudo-code for a callback implementation for saving offsets:
+ * <pre>
+ * {@code
+ * consumer.setRebalanceListener(new RebalanceListener() {
+ * @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(rebalanceConsumer.position(partition));
+ * }
+ *
+ * @Override
+ * public void onPartitionsLost(Collection<TopicPartition> partitions,
RebalanceConsumer rebalanceConsumer) {
+ * // do not need to save the offsets since these partitions are
probably owned by other consumers already
+ * }
+ *
+ * @Override
+ * public void onPartitionsAssigned(Collection<TopicPartition>
partitions, RebalanceConsumer rebalanceConsumer) {
+ * // read the offsets from an external store using some custom code
not described here
+ * for(TopicPartition partition: partitions)
+ * rebalanceConsumer.seek(partition,
readOffsetFromExternalStore(partition));
+ * }
+ * });
+ * consumer.subscribe(List.of("topic-1", "topic-2"));
+ * }
+ * </pre>
+ *
+ * @see RebalanceConsumer
+ * @see ConsumerRebalanceListener
+ */
+public interface RebalanceListener {
+
+ /**
+ * A callback method the user can implement to provide handling of offset
commits to a customized store.
+ * This method will be called during a rebalance operation when the
consumer has to give up some partitions.
+ * The consumer may need to give up some partitions (thus this callback
executed) under the following scenarios:
+ * <ul>
+ * <li>If the consumer assignment changes</li>
+ * <li>If the consumer is being closed ({@link
KafkaConsumer#close(CloseOptions option)})</li>
+ * <li>If the consumer is unsubscribing ({@link
KafkaConsumer#unsubscribe()})</li>
+ * </ul>
+ * It is recommended that offsets should be committed in this callback to
either Kafka or a
+ * custom offset store to prevent duplicate data.
+ * <p>
+ * This callback is always called before re-assigning the partitions.
+ * If the consumer is using the {@link GroupProtocol#CLASSIC} rebalance
protocol:
+ * <ul>
+ * <li>
+ * In eager rebalancing, onPartitionsRevoked will be called with
the full set of assigned partitions as a parameter (all partitions are revoked).
+ * It will be called even if there are no partitions to revoke.
+ * </li>
+ * <li>
+ * In cooperative rebalancing, onPartitionsRevoked will be called
with the set of partitions to revoke,
+ * iff the set is non-empty.
Review Comment:
I'd prefer to see `if and only if` in the javadoc, rather than `iff` which
is not really a proper word. One other instance 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]
