apache/beam · error · ValueError

WriteToKafka(with_headers=True) only supports…

Error message

WriteToKafka(with_headers=True) only supports ByteArraySerializer for key and value.

What it means

WriteToKafka with with_headers=True uses the external Kafka expansion service's header-bearing transform, which requires keys and values to be raw bytes serialized with Kafka's ByteArraySerializer. If key_serializer or value_serializer is anything else, the constructor raises ValueError because the external transform cannot accept other serializers in headers mode.

Solutions

Set key_serializer and value_serializer to WriteToKafka.byte_array_serializer and encode keys/values to bytes yourself.
Disable with_headers if you don't need headers and keep your existing serializers.
Pre-encode records (e.g. JSON/Avro to bytes) before writing when switching to ByteArraySerializer.

Example fix

# before
WriteToKafka(bootstrap_servers, topic, with_headers=True)
# after
WriteToKafka(bootstrap_servers, topic, with_headers=True,
             key_serializer=WriteToKafka.byte_array_serializer,
             value_serializer=WriteToKafka.byte_array_serializer)

Defensive patterns

Strategy: validation

Validate before calling

if with_headers and not (key_serializer == WriteToKafka.byte_array_serializer and
                          value_serializer == WriteToKafka.byte_array_serializer):
    raise ValueError('with_headers=True requires ByteArraySerializer')

Type guard

def supports_headers(key_serializer, value_serializer) -> bool:
    ba = WriteToKafka.byte_array_serializer
    return key_serializer == ba and value_serializer == ba

Try / catch

try:
    _ = WriteToKafka(bootstrap, topic, with_headers=True, key_serializer=ks, value_serializer=vs)
except ValueError as e:
    if 'ByteArraySerializer' in str(e): log.error('Switch serializers or disable with_headers')

Prevention

Encode keys/values to bytes before writing when using headers
Only add with_headers=True after switching serializers
Check serializer names against WriteToKafka.byte_array_serializer

When it happens

Trigger: WriteToKafka(..., with_headers=True, key_serializer='org.apache.kafka.common.serialization.StringSerializer', ...) or any serializer besides WriteToKafka.byte_array_serializer ('org.apache.kafka.common.serialization.ByteArraySerializer').

Common situations: Needing per-record headers (common for tracing/audit metadata) while keeping default String serializers; copying an existing WriteToKafka call and only adding with_headers=True.

Understand the failure class

Background: Conflicting config options: "cannot be used together" — configuration validation errors across open-source libraries — this error's family across 162 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/io/kafka.py:325

    :param producer_config: A dictionary containing the producer configuration.
    :param topic: A Kafka topic name.
    :param key_serializer: A fully-qualified Java class name of a Kafka
        Serializer for the topic's key, e.g.
        'org.apache.kafka.common.serialization.LongSerializer'.
        Default: 'org.apache.kafka.common.serialization.ByteArraySerializer'.
    :param value_serializer: A fully-qualified Java class name of a Kafka
        Serializer for the topic's value, e.g.
        'org.apache.kafka.common.serialization.LongSerializer'.
        Default: 'org.apache.kafka.common.serialization.ByteArraySerializer'.
    :param with_headers: If True, input elements must be beam.Row objects
        containing 'key', 'value', and optional 'headers' fields.
        Only ByteArraySerializer is supported when with_headers=True.
    :param expansion_service: The address (host:port) of the ExpansionService.
    """
    if with_headers and (key_serializer != self.byte_array_serializer or
                         value_serializer != self.byte_array_serializer):
      raise ValueError(
          'WriteToKafka(with_headers=True) only supports '
          'ByteArraySerializer for key and value.')

    urn = self.URN_WITH_HEADERS if with_headers else self.URN
    super().__init__(
        urn,
        NamedTupleBasedPayloadBuilder(
            WriteToKafkaSchema(
                producer_config=producer_config,
                topic=topic,
                key_serializer=key_serializer,
                value_serializer=value_serializer,
            )),
        expansion_service or default_io_expansion_service())

View on GitHub (pinned to 12126d8942)