apache/beam · critical · IllegalStateException

Kafka metadata exists for shard

Error message

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'

What it means

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.

Solutions

1. Use a new/unique sink groupId for the new run (spec.getSinkGroupId() / withSinkGroupId).
2. Delete the stale Kafka metadata for the old groupId/topic partitions.
3. Check for another active pipeline using the same groupId and stop it.
4. If the metadata is known-good (aborted-but-committed case), explicitly clear/override the stored state so the writer has matching state.

Example fix

// before
KafkaIO.write().withSinkGroupId("my-sink-group") // reused across runs
// after
KafkaIO.write().withSinkGroupId("my-sink-group-" + UUID.randomUUID()); // unique per run

String groupId = "sink-" + UUID.randomUUID();

try { write.expand(in); } catch (IllegalStateException e) { if (msg.contains("no stored state")) newGroupId(); }

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• : closing producer after unrecoverable error. The work…
• : consumer thread is interrupted
• consumerPollingTimeout should be > 0.
• Couldn't infer Coder from
• Error while parsing the element

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/189d25136c52faf6. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java:634

          LOG.info("Assigned writer id '{}' to shard {}", writerId, shard);

        } else {
          ShardMetadata metadata = JSON_MAPPER.readValue(committed.metadata(), ShardMetadata.class);

          checkNotNull(metadata.writerId);

          if (writerId == null) {
            // a) This might be a restart of the job from scratch, in which case metatdata
            // should be ignored and overwritten with new one.
            // b) This job might be started with an incorrect group id which is an error.
            // c) There is an extremely small chance that this is a retry of the first bundle
            // where metatdate was committed to Kafka but the bundle results were not committed
            // in Beam, in which case it should be treated as correct metadata.
            // How can we tell these three cases apart? Be safe and throw an exception.
            //
            // We could let users explicitly an option to override the existing metadata.
            //
            throw new IllegalStateException(
                String.format(
                    "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'",
                    shard, spec.getSinkGroupId(), committed.metadata()));
          }

          checkState(
              writerId.equals(metadata.writerId),
              "Writer ids don't match. This is mostly a unintended misuse of groupId('%s')."
                  + "Beam '%s', Kafka '%s'",
              spec.getSinkGroupId(),
              writerId,
              metadata.writerId);

          committedSeqId = metadata.sequenceId;

          checkState(

View on GitHub (pinned to 12126d8942)