...

Public Interfaces

New Broker Configuration

New Client Configuration

...

• GROUP_AUTHORIZATION_FAILED — the client is not authorized
• INVALID_REQUEST — the request is malformed, or the plugin semantically rejected the payload by completing its future with InvalidRequestException
• UNSUPPORTED_VERSION — the coordinator cannot serve this RPC because no topology description plugin is configured
• TOPOLOGY_DESCRIPTION_TOO_LARGE — the serialized topology description exceeds group.streams.topology.description.max.bytes
• UNKNOWN_SERVER_ERROR — the plugin encountered an unexpected error processing the request
• plugin rejected the description because it exceeds the size the plugin is willing to store
• TOPOLOGY_DESCRIPTION_UPDATE_FAILED — the plugin failed to process the request for some other reason; the broker logs the underlying error
• GROUP_ID_GROUP_ID_NOT_FOUND — the specified group does not exist
• NOT_COORDINATOR — the broker is not the coordinator for this group
• COORDINATOR_NOT_AVAILABLE — the coordinator is not available
• COORDINATOR_LOAD_IN_PROGRESS — the coordinator is loading

...

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

import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.Uuid;
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
     * that the payload is semantically invalid;, any other exception is treated as aor with
     * server-side plugin error.{@link org.apache.kafka.common.errors.TopologyDescriptionTooLargeException}
     *
 to signal that the *description is @paramlarger requestContextthan the contextplugin ofis thewilling UpdateStreamsGroupTopologyDescription requestto
     * store. @paramAny groupIdother theexception streamsis groupmapped IDto
     * @param topologyDescriptionId the {@code TOPOLOGY_DESCRIPTION_UPDATE_FAILED}.
     *
     * @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,
                                        StreamsGroupTopologyDescription 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<StreamsGroupTopologyDescription>
        getTopology(AuthorizableRequestContext requestContext,
                    String groupId, Uuid topologyDescriptionId);
}

...

1. The plugin is instantiated at broker startup if group.streams.topology.description.class is configured. A broker without a plugin does not advertise UpdateStreamsGroupTopologyDescription in its ApiVersions response and returns UNSUPPORTED_VERSION for any such request. Because a plugin-less broker never includes a TopologyDescriptionId in heartbeat responses, in normal operation the RPC is only sent against a broker that has a plugin configured.
2. The broker mints a fresh TopologyDescriptionId (a UUID) on group creation and on every topology epoch bump, and persists it on the streams group record as a tagged field. After a successful StreamsGroupHeartbeat, the broker calls requiresTopologyPush on the plugin with the current id. STALE_TOPOLOGY members are skipped. If the plugin returns true, the broker includes the id in the heartbeat response and the client pushes its topology under that id; otherwise the field is omitted. Per the plugin contract requiresTopologyPush should not throw; the broker defensively catches any exception, logs at WARN, and treats the call as false. Beyond the STALE_TOPOLOGY skip the broker does no other gating — deduplication, timeouts, and back-off are the plugin's responsibility (see Plugin Implementation Guidelines). The plugin sees only the opaque id and the group identity; it does not see topology epochs or wall-clock timestamps.

3. On UpdateStreamsGroupTopologyDescription, the broker checks the READ ACL on the group and that a plugin is configured.

   If the serialized description exceeds group.streams.topology.description.max.bytes, the request is rejected with TOPOLOGY_DESCRIPTION_TOO_LARGE. Size is measured at the highest UpdateStreamsGroupTopologyDescription version the broker supports — not the negotiated version — so a pre-validating client should use the same version (discoverable via ApiVersions). The broker then

   The broker does not enforce a size limit on the topology description — it is the plugin's responsibility to decide what size it is willing to store. The broker calls setTopology on the plugin, passing the TopologyDescriptionId from the request unchanged. On success, the response carries NONE; InvalidRequestException from the plugin maps to INVALID_REQUEST, TopologyDescriptionTooLargeException maps to TOPOLOGY_DESCRIPTION_TOO_LARGE, any other exception maps to

   UNKNOWN

   TOPOLOGY_DESCRIPTION_

   SERVER

   UPDATE_

   ERROR

   FAILED, and

   both

   all three are logged at WARN.

4. On DeleteGroups, the broker calls deleteTopology on the plugin only after the group delete has returned no error for that group. deleteTopology failures are logged but do not affect the deletion response. For groups that expire naturally (all members leave), deleteTopology is not called — the plugin expires the topology via a wall-clock TTL (see Plugin Implementation Guidelines).
5. On StreamsGroupDescribe with IncludeTopologyDescription=true, the broker reads the current TopologyDescriptionId from group metadata and calls getTopology on the plugin for each group, after assembling the rest of the response. Calls across groups run in parallel. Authorization is unchanged — the existing DESCRIBE ACL on the GROUP resource covers the topology description. DescribedGroup.ErrorCode is never modified by topology-related outcomes; the TopologyDescriptionStatus field carries the reason when TopologyDescription is null (NOT_REQUESTED, NOT_STORED, or ERROR; the last is also logged at WARN). When no plugin is configured, the broker returns NOT_STORED (the broker does not distinguish "no plugin" from "plugin has nothing stored" on the wire).

...

All other errors (TOPOLOGY_DESCRIPTION_TOO_LARGE, TOPOLOGY_DESCRIPTION_UPDATE_FAILED, INVALID_REQUEST, UNSUPPORTED_VERSION, UNKNOWN_SERVER_ERROR, GROUP_ID_NOT_FOUND, GROUP_AUTHORIZATION_FAILED) clear the pending id and log at WARN. The client does not retry on its own; re-attempts, if any, happen when the broker sends a non-null id on a subsequent heartbeat. See Plugin Implementation Guidelines for when the plugin should re-solicit.

...

• Implement requiresTopologyPush as a non-blocking, efficient call, since it is invoked frequently.
• Expire and clean up stored topology descriptions for inactive groups and superseded ids internally. Incoming requiresTopologyPush calls can be used as an implicit keep-alive.
• Decide and enforce a maximum stored description size. Reject pushes that exceed it by completing the setTopology future with TopologyDescriptionTooLargeException, and stop returning requiresTopologyPush=true for an id once that id has been confirmed too large.
• Avoid concurrent or repetitive pushes: track in-flight pushes per id, back off with an exponential schedule on transient failures, and disable pushes permanently for the id on permanent failures (TOPOLOGY_DESCRIPTION_TOO_LARGE, plugin-semantic INVALID_REQUEST).
• Not request repeated pushes from stable groups that have already pushed their topology under the current id.

...

KIP-714 supports compression (ZStd, LZ4, GZip, Snappy) for telemetry payloads because metrics are pushed repeatedly at high frequency. Topology descriptions are pushed infrequently (only on topology epoch changes) and are the wire payload is expected to fit comfortably under the 350 KB default limit even for complex for typical applications. The Kafka protocol's flexible versions already provide efficient serialization. Adding compression would add complexity without meaningful benefit for this use case.

...