apache/beam · error · IllegalArgumentException

Encoded value for the offset consumer config key

Error message

Encoded value for the offset consumer config key %s was null.

What it means

fromConfigRow decodes the offset_consumer_config map, which overrides consumer settings used for offset/fetch operations. Each entry's encoded value must be a non-null byte[]; a null value is a malformed payload and is rejected with IllegalArgumentException.

Solutions

Provide encoded byte[] values for every offset_consumer_config key.
Remove unneeded keys from the map.
Regenerate the config Row from the source transform using withOffsetConsumerConfigOverrides with valid values.

Example fix

// before
offsetConsumerConfig.put("max.poll.records", null);
// after
offsetConsumerConfig.put("max.poll.records", toByteArray("500"));

Defensive patterns

Strategy: validation

Validate before calling

Map<String, byte[]> occ = configRow.getMap("offset_consumer_config");
if (occ != null) {
  occ.forEach((k, v) -> { if (v == null) throw new IllegalArgumentException("Null encoded offset consumer config value for: " + k); });
}

Type guard

boolean offsetConfigValid(Map<String, byte[]> m) { return m == null || m.values().stream().allMatch(java.util.Objects::nonNull); }

Try / catch

try {
  transform = fromConfigRow(configRow);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("offset consumer config key")) { /* fix or drop the offending key */ }
  throw e;
}

Prevention

Encode all offset consumer override values with toByteArray before serialization.
Never insert null values into offset consumer config maps.
Regenerate payloads with withOffsetConsumerConfigOverrides when in doubt.

When it happens

Trigger: Restoring a transform whose config Row's offset_consumer_config map contains a key mapped to null.

Common situations: Hand-built or buggy serialization of offset consumer overrides; values stripped by tooling; old payloads where nulls were allowed.

Related errors

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

          }
          transform =
              transform.withDynamicRead(
                  org.joda.time.Duration.millis(watchTopicPartitionDuration.toMillis()));
        }

        byte[] timestampPolicyFactory = configRow.getBytes("timestamp_policy_factory");
        if (timestampPolicyFactory != null) {
          transform =
              transform.withTimestampPolicyFactory(
                  (TimestampPolicyFactory) fromByteArray(timestampPolicyFactory));
        }
        Map<String, byte[]> offsetConsumerConfig = configRow.getMap("offset_consumer_config");
        if (offsetConsumerConfig != null) {
          Map<String, Object> updatedOffsetConsumerConfig = new HashMap<>();
          offsetConsumerConfig.forEach(
              (key, dataBytes) -> {
                if (offsetConsumerConfig.get(key) == null) {
                  throw new IllegalArgumentException(
                      "Encoded value for the offset consumer config key " + key + " was null.");
                }
                try {
                  updatedOffsetConsumerConfig.put(
                      key, fromByteArray(offsetConsumerConfig.get(key)));
                } catch (InvalidClassException e) {
                  throw new RuntimeException(e);
                }
              });
          transform = transform.withOffsetConsumerConfigOverrides(updatedOffsetConsumerConfig);
        }

        byte[] checkStopReadinfFn = configRow.getBytes("check_stop_reading_fn");
        if (checkStopReadinfFn != null) {
          transform =
              transform.withCheckStopReadingFn(
                  (SerializableFunction<TopicPartition, Boolean>)
                      fromByteArray(checkStopReadinfFn));

View on GitHub (pinned to 12126d8942)