{"record":{"id":"bc9a87c5e915d810","repo":"apache/kafka","slug":"global-store-must-be-composed-of-a-source-and-a-pr","errorCode":null,"errorMessage":"Global store must be composed of a source and a processor node.","messagePattern":"Global store must be composed of a source and a processor node\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java","lineNumber":265,"sourceCode":"    private StreamsGroupTopologyDescription.Subtopology convertTopologySubtopology(\n            final StreamsGroupDescribeResponseData.TopologyDescriptionSubtopology subtopology) {\n        final Map<String, Set<String>> predecessors = reconstructPredecessors(subtopology.nodes());\n        final List<StreamsGroupTopologyDescription.Node> nodes = subtopology.nodes().stream()\n            .map(node -> convertTopologyNode(node, predecessors))\n            .collect(Collectors.toList());\n        return new StreamsGroupTopologyDescription.Subtopology(subtopology.subtopologyId(), nodes);\n    }\n\n    private StreamsGroupTopologyDescription.GlobalStore convertGlobalStore(\n            final StreamsGroupDescribeResponseData.TopologyDescriptionGlobalStore globalStore) {\n        final List<StreamsGroupDescribeResponseData.TopologyDescriptionNode> pair =\n            List.of(globalStore.source(), globalStore.processor());\n        final Map<String, Set<String>> predecessors = reconstructPredecessors(pair);\n        final StreamsGroupTopologyDescription.Node source = convertTopologyNode(globalStore.source(), predecessors);\n        final StreamsGroupTopologyDescription.Node processor = convertTopologyNode(globalStore.processor(), predecessors);\n        if (!(source instanceof StreamsGroupTopologyDescription.Source)\n                || !(processor instanceof StreamsGroupTopologyDescription.Processor)) {\n            throw new IllegalStateException(\"Global store must be composed of a source and a processor node.\");\n        }\n        return new StreamsGroupTopologyDescription.GlobalStore(\n            (StreamsGroupTopologyDescription.Source) source,\n            (StreamsGroupTopologyDescription.Processor) processor\n        );\n    }\n\n    /**\n     * Reconstructs the predecessor relation from the successor lists carried on the wire. For every node, each of its\n     * successors gains this node as a predecessor.\n     */\n    private Map<String, Set<String>> reconstructPredecessors(\n            final List<StreamsGroupDescribeResponseData.TopologyDescriptionNode> nodes) {\n        final Map<String, Set<String>> predecessors = new HashMap<>();\n        for (final StreamsGroupDescribeResponseData.TopologyDescriptionNode node : nodes) {\n            for (final String successor : node.successors()) {\n                predecessors.computeIfAbsent(successor, ignored -> new HashSet<>()).add(node.name());\n            }","sourceCodeStart":247,"sourceCodeEnd":283,"githubUrl":"https://github.com/apache/kafka/blob/996fb4585aa1bcc8980b0e1b8d6b168b986cd979/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java#L247-L283","documentation":"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.","triggerScenarios":"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.","commonSituations":"Broker/version mismatch; a broker bug emitting malformed global-store entries; an experimental Streams topology format not understood by this client.","solutions":["Align client and broker versions (upgrade the client to match the broker, or downgrade the broker).","Treat this as a broker bug and report it upstream; the client cannot safely guess the structure.","Wrap describeStreamsGroups in a try/catch and fall back to a degraded view (skip topology details)."],"exampleFix":"// before\nStreamsGroupDescription d = admin.describeStreamsGroups(List.of(g)).all().get().get(g);\n\n// after\ntry {\n    StreamsGroupDescription d = admin.describeStreamsGroups(List.of(g)).all().get().get(g);\n} catch (ExecutionException e) {\n    if (e.getCause() instanceof IllegalStateException) {\n        log.warn(\"Malformed streams topology from broker for {}; upgrade client\", g, e);\n    } else throw e;\n}","handlingStrategy":"try-catch","validationCode":"// No client-side pre-validation; the wire payload is opaque. Align versions instead.\nassert clientVersionIsAtLeast(brokerVersion);","typeGuard":null,"tryCatchPattern":"try {\n    return admin.describeStreamsGroups(groups).all().get();\n} catch (ExecutionException e) {\n    if (e.getCause() instanceof IllegalStateException\n        && e.getCause().getMessage().contains(\"Global store\")) {\n        log.warn(\"Malformed global store from broker; upgrade client\", e);\n        return Collections.emptyMap();\n    }\n    throw e;\n}","preventionTips":["Pin client and broker to compatible versions.","Run an integration test exercising describeStreamsGroups with realistic topologies.","Surfaces broker-format bugs early via canary calls in CI."],"tags":["admin-client","streams","wire-protocol","illegal-state","version-mismatch","broker-bug"],"backgroundTag":null,"analyzedSha":"996fb4585aa1bcc8980b0e1b8d6b168b986cd979","analyzedAt":"2026-08-11T22:03:28.655Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}