{"record":{"id":"189d25136c52faf6","repo":"apache/beam","slug":"kafka-metadata-exists-for-shard-s-but-there-is-no-stored","errorCode":null,"errorMessage":"Kafka metadata exists for shard %s, but there is no stored state for it. This mostly indicates groupId '%s' is used else where or in earlier runs. Try another group id. Metadata for this shard on Kafka : '%s'","messagePattern":"Kafka metadata exists for shard (.+?), but there is no stored state for it\\. This mostly indicates groupId '(.+?)' is used else where or in earlier runs\\. Try another group id\\. Metadata for this shard on Kafka : '(.+?)'","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"critical","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java","lineNumber":634,"sourceCode":"          LOG.info(\"Assigned writer id '{}' to shard {}\", writerId, shard);\n\n        } else {\n          ShardMetadata metadata = JSON_MAPPER.readValue(committed.metadata(), ShardMetadata.class);\n\n          checkNotNull(metadata.writerId);\n\n          if (writerId == null) {\n            // a) This might be a restart of the job from scratch, in which case metatdata\n            // should be ignored and overwritten with new one.\n            // b) This job might be started with an incorrect group id which is an error.\n            // c) There is an extremely small chance that this is a retry of the first bundle\n            // where metatdate was committed to Kafka but the bundle results were not committed\n            // in Beam, in which case it should be treated as correct metadata.\n            // How can we tell these three cases apart? Be safe and throw an exception.\n            //\n            // We could let users explicitly an option to override the existing metadata.\n            //\n            throw new IllegalStateException(\n                String.format(\n                    \"Kafka metadata exists for shard %s, but there is no stored state for it. \"\n                        + \"This mostly indicates groupId '%s' is used else where or in earlier runs. \"\n                        + \"Try another group id. Metadata for this shard on Kafka : '%s'\",\n                    shard, spec.getSinkGroupId(), committed.metadata()));\n          }\n\n          checkState(\n              writerId.equals(metadata.writerId),\n              \"Writer ids don't match. This is mostly a unintended misuse of groupId('%s').\"\n                  + \"Beam '%s', Kafka '%s'\",\n              spec.getSinkGroupId(),\n              writerId,\n              metadata.writerId);\n\n          committedSeqId = metadata.sequenceId;\n\n          checkState(","sourceCodeStart":616,"sourceCodeEnd":652,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java#L616-L652","documentation":"KafkaExactlyOnceSink's shard writer detects that Kafka already holds committed metadata for a shard, but the sink has no stored state for that shard. Since the sink cannot distinguish a stale/duplicate groupId run from a legitimately committed-but-aborted bundle, it refuses to proceed to preserve exactly-once guarantees.","triggerScenarios":"initShardWriter (called from processElement) finds metadata for the shard on Kafka whose groupId matches spec.getSinkGroupId() but whose stored state is absent — i.e. the group id was used by another job or an earlier run of the same pipeline.","commonSituations":"Reusing a sink groupId across pipeline runs; running two pipelines concurrently with the same groupId; a previous failed run committed metadata to Kafka but Beam bundle results were never committed; updating the pipeline without changing the group id.","solutions":["Use a new/unique sink groupId for the new run (spec.getSinkGroupId() / withSinkGroupId).","Delete the stale Kafka metadata for the old groupId/topic partitions.","Check for another active pipeline using the same groupId and stop it.","If the metadata is known-good (aborted-but-committed case), explicitly clear/override the stored state so the writer has matching state."],"exampleFix":"// before\nKafkaIO.write().withSinkGroupId(\"my-sink-group\") // reused across runs\n// after\nKafkaIO.write().withSinkGroupId(\"my-sink-group-\" + UUID.randomUUID()); // unique per run","handlingStrategy":"validation","validationCode":"String groupId = \"sink-\" + UUID.randomUUID();","typeGuard":null,"tryCatchPattern":"try { write.expand(in); } catch (IllegalStateException e) { if (msg.contains(\"no stored state\")) newGroupId(); }","preventionTips":["Unique groupIds per run; clean stale metadata"],"tags":["kafka","exactly-once","group-id","state-conflict","beam-io"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}