apache/kafka · error · IllegalStateException

Global store must be composed of a source and a processor no

Error message

Global store must be composed of a source and a processor node.

What it means

Thrown by DescribeStreamsGroupsHandler.convertGlobalStore when a topology global store's source or processor node, after conversion, is not respectively a Source or Processor. A global store in Kafka Streams is structurally a source node feeding a processor that backs a state store; if the broker advertises mismatched node types the client cannot reconstruct the topology and aborts.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java:265

    private StreamsGroupTopologyDescription.Subtopology convertTopologySubtopology(
            final StreamsGroupDescribeResponseData.TopologyDescriptionSubtopology subtopology) {
        final Map<String, Set<String>> predecessors = reconstructPredecessors(subtopology.nodes());
        final List<StreamsGroupTopologyDescription.Node> nodes = subtopology.nodes().stream()
            .map(node -> convertTopologyNode(node, predecessors))
            .collect(Collectors.toList());
        return new StreamsGroupTopologyDescription.Subtopology(subtopology.subtopologyId(), nodes);
    }

    private StreamsGroupTopologyDescription.GlobalStore convertGlobalStore(
            final StreamsGroupDescribeResponseData.TopologyDescriptionGlobalStore globalStore) {
        final List<StreamsGroupDescribeResponseData.TopologyDescriptionNode> pair =
            List.of(globalStore.source(), globalStore.processor());
        final Map<String, Set<String>> predecessors = reconstructPredecessors(pair);
        final StreamsGroupTopologyDescription.Node source = convertTopologyNode(globalStore.source(), predecessors);
        final StreamsGroupTopologyDescription.Node processor = convertTopologyNode(globalStore.processor(), predecessors);
        if (!(source instanceof StreamsGroupTopologyDescription.Source)
                || !(processor instanceof StreamsGroupTopologyDescription.Processor)) {
            throw new IllegalStateException("Global store must be composed of a source and a processor node.");
        }
        return new StreamsGroupTopologyDescription.GlobalStore(
            (StreamsGroupTopologyDescription.Source) source,
            (StreamsGroupTopologyDescription.Processor) processor
        );
    }

    /**
     * Reconstructs the predecessor relation from the successor lists carried on the wire. For every node, each of its
     * successors gains this node as a predecessor.
     */
    private Map<String, Set<String>> reconstructPredecessors(
            final List<StreamsGroupDescribeResponseData.TopologyDescriptionNode> nodes) {
        final Map<String, Set<String>> predecessors = new HashMap<>();
        for (final StreamsGroupDescribeResponseData.TopologyDescriptionNode node : nodes) {
            for (final String successor : node.successors()) {
                predecessors.computeIfAbsent(successor, ignored -> new HashSet<>()).add(node.name());
            }

View on GitHub (pinned to 996fb4585a)

Solutions

1. Align client and broker versions (upgrade the client to match the broker, or downgrade the broker).
2. Treat this as a broker bug and report it upstream; the client cannot safely guess the structure.
3. Wrap describeStreamsGroups in a try/catch and fall back to a degraded view (skip topology details).

Example fix

// before
StreamsGroupDescription d = admin.describeStreamsGroups(List.of(g)).all().get().get(g);

// after
try {
    StreamsGroupDescription d = admin.describeStreamsGroups(List.of(g)).all().get().get(g);
} catch (ExecutionException e) {
    if (e.getCause() instanceof IllegalStateException) {
        log.warn("Malformed streams topology from broker for {}; upgrade client", g, e);
    } else throw e;
}

// No client-side pre-validation; the wire payload is opaque. Align versions instead.
assert clientVersionIsAtLeast(brokerVersion);

try {
    return admin.describeStreamsGroups(groups).all().get();
} catch (ExecutionException e) {
    if (e.getCause() instanceof IllegalStateException
        && e.getCause().getMessage().contains("Global store")) {
        log.warn("Malformed global store from broker; upgrade client", e);
        return Collections.emptyMap();
    }
    throw e;
}

When it happens

Trigger: Receiving a StreamsGroupDescribeResponse whose TopologyDescriptionGlobalStore has a source() whose node type is not NODE_TYPE_SOURCE, or a processor() whose node type is not NODE_TYPE_PROCESSOR.

Common situations: Broker/version mismatch; a broker bug emitting malformed global-store entries; an experimental Streams topology format not understood by this client.

Related errors

• Topology description is missing despite status AVAILABLE
• Unknown topology node type: {nodeType}
• Unknown topology description status id: {id}
• Unknown acknowledge type id: {id}
• Unknown rebalance protocol id: {id}

AI-assisted analysis of apache/kafka@996fb4585a (2026-08-11). Data as JSON: /api/errors/bc9a87c5e915d810. Report an issue: GitHub.