Page History

package org.apache.kafka.coordinator.group.api.streams;

import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.UpdateStreamsGroupTopologyDescriptionRequestData;
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
import java.util.concurrent.CompletableFuture;

/**
 * A broker-side plugin that manages topology descriptions for streams groups.
 *
 * <p>Implementations receive topology descriptions pushed by Kafka Streams clients
 * and can store, forward, or expose them however they see fit.
 *
 * <p>The broker calls {@link #requiresTopologyPush} on every heartbeat to determine
 * whether the client should send its topology description. Implementations that need
 * to consult an external service should kick off that work asynchronously on the
 * first call and return {@code false} until the result is available.
 *
 * <p>Every method takes an {@link AuthorizableRequestContext} as its first argument.
 *
 * <p>Implementations must be thread-safe. {@link #setTopology} may be called
 * concurrently by multiple group members observing the same
 * {@code TopologyDescriptionId} in their heartbeat responses; concurrent
 * calls with the same {@code (groupId, topologyDescriptionId)} pair carry
 * identical data and must be treated as idempotent.
 */
public interface StreamsGroupTopologyDescriptionPlugin extends Configurable, AutoCloseable {

    /**
     * Returns whether the broker should request a topology push from the client.
     *
     * <p>Called on every successful heartbeat. Not called for members in
     * {@code STALE_TOPOLOGY} status. If this method returns {@code true}, the broker
     * includes the {@code topologyDescriptionId} in the heartbeat response.
     *
     * <p>This method should not throw. Failure modes should be handled internally
     * and converted to a return value of {@code false}. The broker defensively
     * catches any exception and treats it as {@code false}, logging at WARN; plugins
     * should not rely on this backstop.
     *
     * <p>This method is on the heartbeat path and must return quickly. Implementations
     * that need to consult an external service should return {@code false} until the
     * result is available.
     *
     * <p>See the KIP's <em>Plugin Implementation Guidelines</em> section for how
     * implementations should handle in-flight tracking, retries, and topology
     * expiration.
     *
     * @param requestContext the context of the heartbeat request
     * @param groupId the streams group ID
     * @param topologyDescriptionId the broker-minted id identifying the current topology
     *        version for this group; opaque to the plugin
     * @return true if the broker should request a topology push from the client
     */
    boolean requiresTopologyPush(AuthorizableRequestContext requestContext,
                                 String groupId, Uuid topologyDescriptionId);

    /**
     * Called when a client sends a topology description for a streams group.
     * This method may be called concurrently by multiple members of the same group;
     * all calls for the same (groupId, topologyDescriptionId) carry identical data.
     *
     * <p>The returned future completes when the topology has been persisted or
     * forwarded. Complete it exceptionally with
     * {@link org.apache.kafka.common.errors.InvalidRequestException} to signal that
     * the payload is semantically invalid; any other exception is treated as a
     * server-side plugin error.
     *
     * @param requestContext the context of the UpdateStreamsGroupTopologyDescription request
     * @param groupId the streams group ID
     * @param topologyDescriptionId the id this push is tagged with, as carried in the
     *        heartbeat response that asked for it; opaque to the plugin
     * @param description the topology description
     * @return a future that completes when the operation is done
     */
    CompletableFuture<Void> setTopology(AuthorizableRequestContext requestContext,
                                        String groupId, Uuid topologyDescriptionId,
                                        UpdateStreamsGroupTopologyDescriptionRequestData.TopologyDescriptionStreamsGroupTopologyDescription description);

      /**
     * Called when a group is explicitly deleted via DeleteGroups. Removes any topology
     * description stored for this group.
     *
     * <p>The returned future completes when the deletion has been processed.
     * If it completes exceptionally, the broker logs the error; the outcome does not
     * affect the DeleteGroups response returned to the caller.
     *
     * @param requestContext the context of the DeleteGroups request
     * @param groupId the streams group ID
     * @return a future that completes when the operation is done
     */
    CompletableFuture<Void> deleteTopology(AuthorizableRequestContext requestContext, String groupId);

    /**
     * Called to retrieve the stored topology description for a group. This is invoked
     * by the broker when a client calls StreamsGroupDescribe with
     * {@code IncludeTopologyDescription=true}.
     *
     * <p>Returns a future that resolves to the stored topology description for the
     * given {@code (groupId, topologyDescriptionId)} pair, or to {@code null} if no
     * topology is stored (e.g. no push has succeeded yet, or the stored description
     * is tagged with a different id). If the future completes exceptionally, the
     * plugin signals a read error for this group.
     *
     * @param requestContext the context of the StreamsGroupDescribe request
     * @param groupId the streams group ID
     * @param topologyDescriptionId the id of the topology version the caller is
     *        asking about; opaque to the plugin
     * @return a future resolving to the stored topology description, or null if none
     */
    CompletableFuture<UpdateStreamsGroupTopologyDescriptionRequestData.TopologyDescription>CompletableFuture<StreamsGroupTopologyDescription>
        getTopology(AuthorizableRequestContext requestContext,
                    String groupId, Uuid topologyDescriptionId);
}

package org.apache.kafka.clients.admin;

public class StreamsGroupTopologyDescription {
    public Collection<Subtopology> subtopologies();
    public Collection<GlobalStore> globalStores();

    public static class Subtopology {
        public String id();
        public Collection<Node> nodes();
    }

    public interface Node {
        String name();
        Set<String> predecessors();
        Set<String> successors();
    }

    public static class Source implements Node {
        public Set<String> topics();
    }

    public static class Processor implements Node {
        public Set<String> stores();
    }

    public static class Sink implements Node {
        public String topic();
    }

    public static class GlobalStore {
        public Source source();
        public Processor processor();
    }
}