{"record":{"id":"55e6e220fd1fc951","repo":"apache/beam","slug":"s-is-not-a-runner-known-to-be-compatible-with-kafka-exactly","errorCode":null,"errorMessage":"%s is not a runner known to be compatible with Kafka exactly-once sink. This implementation of exactly-once sink relies on specific checkpoint guarantees. Only the runners with known to have compatible checkpoint semantics are allowed.","messagePattern":"(.+?) is not a runner known to be compatible with Kafka exactly-once sink\\. This implementation of exactly-once sink relies on specific checkpoint guarantees\\. Only the runners with known to have compatible checkpoint semantics are allowed\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java","lineNumber":3597,"sourceCode":"                pCollectionTuple\n                    .get(BadRecordRouter.BAD_RECORD_TAG)\n                    .setCoder(BadRecord.getCoder(input.getPipeline())));\n      }\n      return PDone.in(input.getPipeline());\n    }\n\n    @Override\n    public void validate(@Nullable PipelineOptions options) {\n      Preconditions.checkStateNotNull(options);\n      if (isEOS()) {\n        String runner = options.getRunner().getName();\n        if (\"org.apache.beam.runners.direct.DirectRunner\".equals(runner)\n            || runner.startsWith(\"org.apache.beam.runners.dataflow.\")\n            || runner.startsWith(\"org.apache.beam.runners.spark.\")\n            || runner.startsWith(\"org.apache.beam.runners.flink.\")) {\n          return;\n        }\n        throw new UnsupportedOperationException(\n            runner\n                + \" is not a runner known to be compatible with Kafka exactly-once sink. This\"\n                + \" implementation of exactly-once sink relies on specific checkpoint guarantees.\"\n                + \" Only the runners with known to have compatible checkpoint semantics are\"\n                + \" allowed.\");\n      }\n    }\n\n    // set config defaults\n    private static final Map<String, Object> DEFAULT_PRODUCER_PROPERTIES =\n        ImmutableMap.of(ProducerConfig.RETRIES_CONFIG, 3);\n\n    /** A set of properties that are not required or don't make sense for our producer. */\n    private static final Map<String, String> IGNORED_PRODUCER_PROPERTIES =\n        ImmutableMap.of(\n            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, \"Use withKeySerializer instead\",\n            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, \"Use withValueSerializer instead\");\n","sourceCodeStart":3579,"sourceCodeEnd":3615,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L3579-L3615","documentation":"The Kafka exactly-once sink relies on runner checkpointing semantics to commit produced records atomically. Before expansion, KafkaIO verifies the configured runner is one with known-compatible checkpointing (Direct, Dataflow, Spark, Flink); any other runner raises this UnsupportedOperationException.","triggerScenarios":"Using KafkaIO.writeRecords().withEOS(...) (exactly-once semantics) on a runner other than Direct/Dataflow/Spark/Flink — e.g. a custom or experimental runner, or a runner class name that doesn't start with the recognized package prefixes.","commonSituations":"Running exactly-once Kafka writes on a new/alternative runner (Samza, custom portable runner), or setting --runner to a subclass in an unrecognized package.","solutions":["Run on a supported runner (Direct for tests, Dataflow, Spark, or Flink).","Disable exactly-once sink (drop withEOS / use ACK-based or at-least-once sink) if the runner can't change.","If you believe your runner has compatible checkpoint semantics, use a non-EOS sink or contact the Beam community; there is no override flag."],"exampleFix":"// before\n--runner=MyCustomRunner  (with .withEOS(30, \"test\"))\n// after\n--runner=FlinkRunner  (with .withEOS(30, \"test\"))","handlingStrategy":"validation","validationCode":"String runner = options.getRunner().getName();\nboolean eosCompatible = runner.equals(\"org.apache.beam.runners.direct.DirectRunner\")\n    || runner.startsWith(\"org.apache.beam.runners.dataflow.\")\n    || runner.startsWith(\"org.apache.beam.runners.spark.\")\n    || runner.startsWith(\"org.apache.beam.runners.flink.\");\nif (usingEosSink && !eosCompatible) {\n  throw new IllegalArgumentException(\"Exactly-once Kafka sink requires Direct/Dataflow/Spark/Flink\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  pipeline.run().waitUntilFinish();\n} catch (UnsupportedOperationException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"exactly-once sink\")) {\n    // reconfigure write without withEOS or switch runner\n  }\n  throw e;\n}","preventionTips":["Only enable withEOS on Direct/Dataflow/Spark/Flink runners","Gate EOS configuration behind a pipeline-option check","For unsupported runners use the at-least-once sink with idempotent consumers"],"tags":["java","kafka","runner","exactly-once"],"backgroundTag":"unsupported-operation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}