Page History

package org.apache.kafka.clients.consumer;

/**
 * {@inheritDoc}
 * @deprecated
 * @see RebalanceListener
 */
@Deprecated(since = "4.4", forRemoval = true)
public interface ConsumerRebalanceListener extends RebalanceListener {
   
    // Other methods stay unchanged.

	/**
     * Called after partitions have been assigned to this consumer.
     *
     * <p>The default implementation delegates to
     * {@link #onPartitionsAssigned(Collection)}. Override this method
     * instead of the one-arg variant when the callback needs to interact
     * with the consumer (commit offsets, seek, pause/resume).
     *
     * <p>When this method is overridden, the one-arg
     * {@code onPartitionsAssigned(Collection)} is not called unless the
     * override explicitly invokes it.
     *
     * @param partitions the newly assigned partitions
     * @param consumer   a restricted view of the consumer; supports seek,
     *                   commit, pause/resume, and read-only queries
     */
    @Override
    default void onPartitionsAssigned(Collection<TopicPartition> partitions, RebalanceConsumer consumer) {
        onPartitionsAssigned(partitions);
    }

    /**
     * Called before partitions are revoked from this consumer during a
     * cooperative rebalance, or before all partitions are revoked during
     * an eager rebalance.
     *
     * <p>The default implementation delegates to
     * {@link #onPartitionsRevoked(Collection)}.
     *
     * @param partitions the partitions being revoked
     * @param consumer   a restricted view of the consumer
     */
    @Override
    default void onPartitionsRevoked(Collection<TopicPartition> partitions, RebalanceConsumer consumer) {
        onPartitionsRevoked(partitions);
    }

    /**
     * Called when partitions have been lost. Unlike revocation, the consumer
     * no longer owns these partitions and commits will fail.
     *
     * <p>The default implementation delegates to {@link #onPartitionsLost(Collection)}.
     *
     * @param partitions the partitions that were lost
     * @param consumer   a restricted view of the consumer
     */
    @Override
    default void onPartitionsLost(Collection<TopicPartition> partitions, RebalanceConsumer consumer) {
        onPartitionsLost(partitions);
    }
}

package org.apache.kafka.clients.consumer;

/**
 * A restricted view of a {@link Consumer} passed to {@link ConsumerRebalanceListener} methods.
 * This object is only valid for the duration of the rebalance callback. Any use outside of this scope,
 * will result in an {@link IllegalStateException}.
 *
 * <p>This interface exposes only operations that are safe and meaningful
 * during partition rebalancing:
 *
 * <ul>
 *   <li><b>Offset management:</b> {@code commitSync}, {@code commitAsync},
 *       {@code committed}, {@code position}</li>
 *   <li><b>Seek:</b> {@code seek}, {@code seekToBeginning},
 *       {@code seekToEnd}</li>
 *   <li><b>Partition state:</b> {@code assignment}, {@code paused},
 *       {@code pause}, {@code resume}</li>
 *   <li><b>Read-only queries:</b> {@code partitionsFor}, {@code listTopics},
 *       {@code offsetsForTimes}, {@code beginningOffsets},
 *       {@code endOffsets}, {@code metrics}, {@code clientInstanceId}, {@code currentLag}, {@code groupMetadata}</li>
 * </ul>
 *
 * <p>Operations that are dangerous, inapplicable during a rebalance are not present
 * on this interface:
 *
 * <ul>
 *   <li>{@code poll()} - would cause re-entrant polling</li>
 *   <li>{@code close()} - would terminate the consumer mid-rebalance</li>
 *   <li>{@code subscribe()}, {@code unsubscribe()} - would corrupt
 *       subscription state during rebalance</li>
 *   <li>{@code registerMetricForSubscription}, {@code unregisterMetricForSubscription} - metric
 *       registration does not belong during rebalance events</li>
 *   <li>{@code assign()} - conflicts with group-managed assignment</li>
 *   <li>{@code wakeup()} - would interrupt the rebalance</li>
 *   <li>{@code enforceRebalance()} - would trigger re-entrant rebalance</li>
 * </ul>
 */
public interface RebalanceConsumer {

    // --- Offset management ---

    void commitSync();
    void commitSync(Duration timeout);
    void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);
    void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets, Duration timeout);
    void commitAsync();
    void commitAsync(OffsetCommitCallback callback);
    void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets,
                     OffsetCommitCallback callback);

    Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions);
    Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions,
                                                      Duration timeout);
    long position(TopicPartition partition);
    long position(TopicPartition partition, Duration timeout);

    // --- Seek ---

    void seek(TopicPartition partition, long offset);
    void seek(TopicPartition partition, OffsetAndMetadata offsetAndMetadata);
    void seekToBeginning(Collection<TopicPartition> partitions);
    void seekToEnd(Collection<TopicPartition> partitions);

    // --- Partition state ---

    Set<TopicPartition> assignment();
    void pause(Collection<TopicPartition> partitions);
    void resume(Collection<TopicPartition> partitions);
    Set<TopicPartition> paused();

    // --- Read-only queries ---

	Uuid clientInstanceId(Duration timeout);
    Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions);
    Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions,
                                                Duration timeout);
    Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions);
    Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions,
                                         Duration timeout);
    Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(
            Map<TopicPartition, Long> timestampsToSearch);
    Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(
            Map<TopicPartition, Long> timestampsToSearch, Duration timeout);
    List<PartitionInfo> partitionsFor(String topic);
    List<PartitionInfo> partitionsFor(String topic, Duration timeout);
    Map<String, List<PartitionInfo>> listTopics();
    Map<String, List<PartitionInfo>> listTopics(Duration timeout);
	OptionalLong currentLag(TopicPartition topicPartition);
    ConsumerGroupMetadata groupMetadata();
}