apache/seatunnel · error · CassandraConnectorException

ADD_BATCH_DATA_FAILED

ADD_BATCH_DATA_FAILED

Error message

ADD_BATCH_DATA_FAILED

What it means

CassandraSinkWriter.addIntoBatch builds a BoundStatement from the SeaTunnel row and executes it (sync, or async added to completionStages). Any exception in that per-row conversion/execution path is wrapped as ADD_BATCH_DATA_FAILED. Note async write failures may surface later at flush, but binding/type-conversion errors throw here.

Source

Thrown at seatunnel-connectors-v2/connector-cassandra/src/main/java/org/apache/seatunnel/connectors/seatunnel/cassandra/sink/CassandraSinkWriter.java:131

        }
    }

    private void addIntoBatch(SeaTunnelRow row, BoundStatement boundStatement) {
        try {
            for (int i = 0; i < cassandraParameters.getFields().size(); i++) {
                String fieldName = cassandraParameters.getFields().get(i);
                DataType dataType = tableSchema.get(i).getType();
                Object fieldValue = row.getField(seaTunnelRowType.indexOf(fieldName));
                boundStatement =
                        TypeConvertUtil.reconvertAndInject(boundStatement, i, dataType, fieldValue);
            }
            if (cassandraParameters.getAsyncWrite()) {
                completionStages.add(session.executeAsync(boundStatement));
            } else {
                boundStatementList.add(boundStatement);
            }
        } catch (Exception e) {
            throw new CassandraConnectorException(
                    CassandraConnectorErrorCode.ADD_BATCH_DATA_FAILED, e);
        }
    }

    private String initPrepareCQL() {
        String[] placeholder = new String[cassandraParameters.getFields().size()];
        Arrays.fill(placeholder, "?");
        return String.format(
                "INSERT INTO %s (%s) VALUES (%s)",
                cassandraParameters.getTable(),
                String.join(",", cassandraParameters.getFields()),
                String.join(",", placeholder));
    }

    @Override
    public void close() throws IOException {
        flush();
        try {

View on GitHub (pinned to cf67b549a7)

Solutions

  1. Align the sink's `fields` list and SeaTunnel schema types exactly with the Cassandra table columns (DESCRIBE TABLE and compare types).
  2. Check getCause() of this exception for the driver's specific error (InvalidQuery, CodecNotFound, NoNodeAvailable) and address that.
  3. Verify the table still exists and the prepared CQL matches the configured fields after any schema migration.
  4. If timeouts/no-node errors occur, check cluster health, consistency level settings, and network between the SeaTunnel workers and Cassandra.

Example fix

// before: row has 3 values, fields has 2 -> bind fails
fields = ["id", "name"]
// after: include all columns written by upstream
fields = ["id", "name", "age"]
Defensive patterns

Strategy: validation

Validate before calling

// assert row arity and types match fields config before writing
assert row.size() == fields.size();

Try / catch

// catch per-write failures and inspect the driver cause
try {
    writer.write(row);
} catch (CassandraConnectorException e) {
    log.error("bind/execute failed: {}", e.getCause(), e);
    throw e;
}

Prevention

When it happens

Trigger: Exception while creating/binding the prepared statement for a row — e.g. row field count or types don't match the configured fields, null handling issues, or synchronous execute throwing driver errors (invalid query, timeout, no nodes available).

Common situations: SeaTunnel schema field types incompatible with Cassandra column types (e.g. writing a string into an int column); number of configured fields differs from row arity; keyspace/table dropped mid-run; cluster node down.

Related errors

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