apache/beam · error · RuntimeException

Upgrading KafkaIO write transforms that have…

Error message

Upgrading KafkaIO write transforms that have `withBadRecordErrorHandler` property set is not supported yet.

What it means

During Beam pipeline upgrade/translation, KafkaIO write transforms are converted to a config Row via toConfigRow. If the transform has a custom bad-record error handler attached (withBadRecordErrorHandler) that is not the default handler, the upgrade path cannot represent it, so translation fails with this RuntimeException.

Solutions

Remove the custom withBadRecordErrorHandler from the KafkaIO write transform before upgrading, or temporarily use the default ErrorHandler.DefaultErrorHandler
Implement equivalent bad-record handling downstream of the sink (e.g., a separate dead-letter write path) instead of via withBadRecordErrorHandler
Wait for / upgrade to a Beam version that supports translating withBadRecordErrorHandler for KafkaIO writes

Example fix

// before
KafkaIO.<KafkaRecord<Void, byte[]>>writeRecords()
    .withBootstrapServers(servers)
    .withTopic(topic)
    .withBadRecordErrorHandler(customErrorHandler);
// after
KafkaIO.<KafkaRecord<Void, byte[]>>writeRecords()
    .withBootstrapServers(servers)
    .withTopic(topic);

Defensive patterns

Strategy: validation

Validate before calling

if (transform.getBadRecordErrorHandler() != null
    && !(transform.getBadRecordErrorHandler() instanceof ErrorHandler.DefaultErrorHandler)) {
  throw new IllegalStateException("Detach custom bad-record handler before upgrading KafkaIO write transform");
}

Try / catch

try {
  row = translation.toConfigRow(transform);
} catch (RuntimeException e) {
  if (e.getMessage().contains("withBadRecordErrorHandler")) {
    // re-plan without custom handler or abort upgrade with clear user guidance
  }
}

Prevention

Avoid withBadRecordErrorHandler with custom handlers on KafkaIO writes if you plan to use the upgrade/managed path
Keep dead-letter handling as a separate downstream transform
Pin Beam version and check release notes for translation support before upgrading

When it happens

Trigger: Running pipeline upgrade (toConfigRow/row) on a KafkaIO<KafkaRecord<Void,V>>.writeRecords() transform where .withBadRecordErrorHandler(handler) was called with a custom ErrorHandler implementation (anything other than ErrorHandler.DefaultErrorHandler).

Common situations: Migrating an existing Beam Kafka sink pipeline to the newer KafkaIO managed/upgrade representation when the pipeline was written with custom dead-letter error handling for producer failures.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

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

Appendix: source

Thrown at sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java:580

      if (writeRecordsTransform.getConsumerFactoryFn() != null) {
        fieldValues.put(
            "consumer_factory_fn", toByteArray(writeRecordsTransform.getConsumerFactoryFn()));
      }

      if (writeRecordsTransform.getProducerConfig().size() > 0) {
        Map<String, byte[]> producerConfigMap = new HashMap<>();
        writeRecordsTransform
            .getProducerConfig()
            .forEach(
                (key, value) -> {
                  producerConfigMap.put((String) key, toByteArray(value));
                });
        fieldValues.put("producer_config", producerConfigMap);
      }
      if (writeRecordsTransform.getBadRecordErrorHandler() != null
          && !(writeRecordsTransform.getBadRecordErrorHandler()
              instanceof ErrorHandler.DefaultErrorHandler)) {
        throw new RuntimeException(
            "Upgrading KafkaIO write transforms that have `withBadRecordErrorHandler` property set"
                + " is not supported yet.");
      }

      return Row.withSchema(schema).withFieldValues(fieldValues).build();
    }

    @Override
    public Write<?, ?> fromConfigRow(Row configRow, PipelineOptions options) {
      try {
        Write<?, ?> transform = KafkaIO.write();

        String bootstrapServers = configRow.getString("bootstrap_servers");
        if (bootstrapServers != null) {
          transform = transform.withBootstrapServers(bootstrapServers);
        }
        String topic = configRow.getValue("topic");
        if (topic != null) {

View on GitHub (pinned to 12126d8942)