apache/beam · error · ValueError

producer_properties is no longer supported and will be…

Error message

producer_properties is no longer supported and will be removed in a future release.

What it means

The EFO/aggregated WriteToKinesis transform no longer accepts producer_properties; that parameter was removed as the sink moved its configuration into named structured options (aggregation settings, etc.). Passing a non-None producer_properties raises ValueError immediately so users migrate rather than silently losing settings.

Solutions

Remove the producer_properties argument entirely.
Map old producer settings to the named constructor arguments (e.g. aggregation_enabled, aggregation_max_buffered_records, aggregation_max_buffered_time, aggregation_shard_refresh_interval).
Keep any needed producer config in the Kinesis client configuration provided to the expansion service instead.

Example fix

# before
WriteToKinesis(stream_name, region, producer_properties={'CollectionMaxRecords': 500})
# after
WriteToKinesis(stream_name, region, aggregation_enabled=True, aggregation_max_buffered_records=500)

Defensive patterns

Strategy: fallback

Validate before calling

if producer_properties is not None:
    raise ValueError('producer_properties removed; use named aggregation options')

Type guard

def uses_removed_kinesis_options(kwargs) -> bool:
    return 'producer_properties' in kwargs and kwargs['producer_properties'] is not None

Try / catch

try:
    _ = WriteToKinesis(stream, region, producer_properties=props)
except ValueError as e:
    if 'producer_properties' in str(e): log.error('Migrate to named aggregation_* arguments')

Prevention

Drop producer_properties when upgrading Beam Kinesis IO
Translate old producer config keys to aggregation_* constructor arguments
Check the IO changelog during Beam version upgrades

When it happens

Trigger: Constructing apache_beam.io.kinesis.WriteToKinesis(..., producer_properties={...}) with any dict/value, typically ported from older Beam code or the Kinesis IO of a different SDK.

Common situations: Upgrading apache-beam after producer_properties was deprecated and removed; migrating code from the legacy Kinesis IO or from Beam Java examples using producer properties.

Understand the failure class

Background: "is deprecated and will be removed" — deprecation warnings for old API names, keywords, and options, and how to migrate before the removal release — this error's family across 29 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/io/kinesis.py:172

    :param region: AWS region. Example: 'us-east-1'.
    :param service_endpoint: Kinesis service endpoint
    :param verify_certificate: Enable or disable certificate verification.
        Never set to False on production. True by default.
    :param partition_key: Specify default partition key.
    :param producer_properties: (Deprecated) This option no longer is available
        since the AWS IOs upgraded to v2. Trying to set it will lead to an
        error. For more info, see https://github.com/apache/beam/issues/33430.
    :param expansion_service: The address (host:port) of the ExpansionService.
    :param aggregation_enabled: Enable or disable aggregation.
    :param aggregation_max_bytes: Maximum number of bytes to buffer before
        sending a batch of records. Defaults to 51200.
    :param aggregation_max_buffered_time: Maximum time(millisecond) to buffer
        records before sending a batch of records. Defaults to 100.
    :param aggregation_shard_refresh_interval: Interval in minutes to refresh
        the shard map. Defaults to 2.
    """
    if producer_properties is not None:
      raise ValueError(
          'producer_properties is no longer supported and will be removed ' +
          'in a future release.')
    super().__init__(
        self.URN,
        NamedTupleBasedPayloadBuilder(
            WriteToKinesisSchema(
                stream_name=stream_name,
                aws_access_key=aws_access_key,
                aws_secret_key=aws_secret_key,
                region=region,
                partition_key=partition_key,
                service_endpoint=service_endpoint,
                verify_certificate=verify_certificate,
                aggregation_enabled=aggregation_enabled,
                aggregation_max_bytes=aggregation_max_bytes,
                aggregation_max_buffered_time=aggregation_max_buffered_time,
                aggregation_shard_refresh_interval=
                aggregation_shard_refresh_interval,

View on GitHub (pinned to 12126d8942)