apache/seatunnel · error · PulsarConnectorException

CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT

CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT

Error message

Partition key field not found: %s, rowType: %s

What it means

When partition-key-based routing is enabled via partitionKeyFields, each listed field must exist in the SeaTunnelRowType schema. If any name in the config does not match a row field, the writer throws this error at initialization (getPartitionKeyFields) with the offending field and full rowType listed.

Source

Thrown at seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java:329

                row -> {
                    Object[] keyFields = new Object[keyFieldIndexArr.length];
                    for (int i = 0; i < keyFieldIndexArr.length; i++) {
                        keyFields[i] = row.getField(keyFieldIndexArr[i]);
                    }
                    return new SeaTunnelRow(keyFields);
                };
        return row -> keySerializationSchema.serialize(keyDataExtractor.apply(row));
    }

    private List<String> getPartitionKeyFields(
            ReadonlyConfig pluginConfig, SeaTunnelRowType seaTunnelRowType) {
        if (pluginConfig.getOptional(PulsarSinkOptions.PARTITION_KEY_FIELDS).isPresent()) {
            List<String> partitionKeyFields =
                    pluginConfig.get(PulsarSinkOptions.PARTITION_KEY_FIELDS);
            List<String> rowTypeFieldNames = Arrays.asList(seaTunnelRowType.getFieldNames());
            for (String partitionKeyField : partitionKeyFields) {
                if (!rowTypeFieldNames.contains(partitionKeyField)) {
                    throw new PulsarConnectorException(
                            CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
                            String.format(
                                    "Partition key field not found: %s, rowType: %s",
                                    partitionKeyField, rowTypeFieldNames));
                }
            }
            return partitionKeyFields;
        }
        return Collections.emptyList();
    }

    private TransactionImpl createTransaction() {
        try {
            return (TransactionImpl)
                    PulsarConfigUtil.getTransaction(pulsarClient, transactionTimeout);
        } catch (Exception e) {
            throw new PulsarConnectorException(
                    PulsarConnectorErrorCode.CREATE_TRANSACTION_FAILED,

View on GitHub (pinned to cf67b549a7)

Solutions

1. Correct partition_key_fields to exactly match field names in the upstream schema (case-sensitive).
2. Inspect the rowType printed in the message and align the config with it.
3. If the upstream schema can drift, adjust the transform to guarantee the partition key column exists or use a different routing mode.

Example fix

// before
sink {
  Pulsar {
    partition_key_fields = "userId"
  }
}
// after (schema has 'user_id')
sink {
  Pulsar {
    partition_key_fields = "user_id"
  }
}

List<String> rowFields = Arrays.asList(rowType.getFieldNames());
for (String f : partitionKeyFields) {
    if (!rowFields.contains(f)) {
        throw new IllegalArgumentException("partition_key_field not in schema: " + f);
    }
}

boolean allFieldsExist = (List<String> keys, SeaTunnelRowType t) -> Arrays.asList(t.getFieldNames()).containsAll(keys);

try {
    writer = new PulsarSinkWriter(pluginConfig, rowType);
} catch (PulsarConnectorException e) {
    if (e.getMessage().startsWith("Partition key field not found")) {
        log.error("Fix partition_key_fields to match schema: {}", rowType.getFieldNames());
    }
    throw e;
}

When it happens

Trigger: Configuring PulsarSinkOptions.PARTITION_KEY_FIELDS with a field name not present in seaTunnelRowType.getFieldNames(); getPartitionKeyFields runs during writer construction.

Common situations: Field renamed upstream (source schema change) while sink config kept the old name; case mismatch ('UserId' vs 'userId'); typo; schema derived dynamically so expected column missing.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT
• Failed to close Pulsar client after aborting transactions.
• CommonErrorCode.ILLEGAL_ARGUMENT
• CommonErrorCode.ILLEGAL_ARGUMENT
• Failed to close Pulsar sink writer.

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/73abf708fcb9a382. Report an issue: GitHub.