apache/beam · error · RuntimeException

Upgrading KafkaIO read transforms that have…

Error message

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

What it means

During pipeline upgrade, KafkaIOTranslation.toConfigRow serializes a KafkaIO read transform into a config Row. The `withBadRecordErrorHandler` property has no serialized representation yet, so if the transform sets it, the translator refuses to encode the transform and throws, preventing the pipeline from being upgraded/portable-ized.

Solutions

1. Remove the withBadRecordErrorHandler call from the transform before upgrading.
2. Handle bad records manually (e.g. wrap the downstream DoFn with error capture, or use a separate DLQ sink) instead of relying on the built-in handler.
3. Upgrade the pipeline on a Beam version where the handler is unset, then re-add the property after translation support lands.
4. Track Beam's KafkaIOTranslation changes; once supported, re-add withBadRecordErrorHandler.

Example fix

// before
KafkaIO.<byte[], byte[]>read().withBadRecordErrorHandler(handler)
// after
KafkaIO.<byte[], byte[]>read() // no bad record handler during upgrade

if (transform.getBadRecordErrorHandler() != null) {
  throw new IllegalStateException("Remove withBadRecordErrorHandler before pipeline upgrade/translation.");
}

When it happens

Trigger: Calling KafkaIO.Read.create()...withBadRecordErrorHandler(handler) and then running the pipeline through the upgrade path (row()/toConfigRow payload translation, e.g. legacy-to-portable upgrade or ReadRegistrar payload translation).

Common situations: Pipelines that use bad-record error handling on Kafka reads being migrated between Beam versions or runner environments; a developer added withBadRecordErrorHandler to an existing pipeline and then tried to upgrade/translate it.

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

• Upgrading KafkaIO write transforms that have…
• Couldn't infer Coder from
• Attempted to encode null for non-nullable field
• Azure credential provider type
• bad base type

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/6da61b10cfb815e8. 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:219

            .forEach(
                (key, val) -> {
                  offsetConsumerConfigMap.put(key, toByteArray(val));
                });
        fieldValues.put("offset_consumer_config", offsetConsumerConfigMap);
      }
      if (transform.getKeyDeserializerProvider() != null) {
        fieldValues.put(
            "key_deserializer_provider", toByteArray(transform.getKeyDeserializerProvider()));
      }
      if (transform.getValueDeserializerProvider() != null) {
        fieldValues.put(
            "value_deserializer_provider", toByteArray(transform.getValueDeserializerProvider()));
      }
      if (transform.getCheckStopReadingFn() != null) {
        fieldValues.put("check_stop_reading_fn", toByteArray(transform.getCheckStopReadingFn()));
      }
      if (transform.getBadRecordErrorHandler() != null) {
        throw new RuntimeException(
            "Upgrading KafkaIO read transforms that have `withBadRecordErrorHandler` property set"
                + " is not supported yet.");
      }
      if (transform.getLogTopicVerification() != null) {
        fieldValues.put("log_topic_verification", transform.getLogTopicVerification());
      }

      fieldValues.put("redistribute", transform.isRedistributed());
      fieldValues.put("redistribute_num_keys", transform.getRedistributeNumKeys());
      fieldValues.put("allows_duplicates", transform.isAllowDuplicates());
      if (transform.getOffsetDeduplication() != null) {
        fieldValues.put("offset_deduplication", transform.getOffsetDeduplication());
      }
      if (transform.getRedistributeByRecordKey() != null) {
        fieldValues.put("redistribute_by_record_key", transform.getRedistributeByRecordKey());
      }
      return Row.withSchema(schema).withFieldValues(fieldValues).build();
    }

View on GitHub (pinned to 12126d8942)