apache/seatunnel · error · IllegalArgumentException

source position sourceId must not be blank

Error message

source position sourceId must not be blank

What it means

SqliteSourcePositionStore.save() validates that the EdgeSourcePosition carries a non-blank sourceId before persisting the position row. A null or whitespace-only sourceId would create an unusable position record, so the store rejects it with this IllegalArgumentException.

Solutions

  1. Set a non-blank sourceId on the EdgeSourcePosition before saving (e.g. position.setSourceId("my-source") or the configured source name).
  2. Check the source configuration so the id used to build the position is populated.
  3. Validate the position object in caller code before passing it to the store.
  4. If the position came from a serialized record, verify the source produced complete position data.

Example fix

// before
store.save(new EdgeSourcePosition(null, partition, offset, 0));
// after
store.save(new EdgeSourcePosition("kafka-source-1", partition, offset, 0));
Defensive patterns

Strategy: validation

Validate before calling

if (position.getSourceId() == null || position.getSourceId().trim().isEmpty()) { throw new IllegalStateException("sourceId required before save"); } store.save(position);

Type guard

boolean hasSourceId(EdgeSourcePosition p) { return p.getSourceId() != null && !p.getSourceId().trim().isEmpty(); }

Try / catch

try { store.save(position); } catch (IllegalArgumentException e) { if (e.getMessage().contains("sourceId must not be blank")) { log.error("Position missing sourceId: {}", position); } throw e; }

Prevention

When it happens

Trigger: Calling positionStore.save(position) with EdgeSourcePosition whose getSourceId() is null or only whitespace, e.g. built via a builder without setSourceId or constructed from a config where the source id option was missing.

Common situations: A source connector was configured without an explicit id/name; deserializing a position record from an external store with missing fields; programmatic use of the store with a partially populated EdgeSourcePosition.

Understand the failure class

Background: "must not be empty", "cannot be empty" — required-field validation errors across open-source libraries — this error's family across 41 libraries.

Related errors

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

Appendix: source

Thrown at seatunnel-edge-agent/seatunnel-edge-agent-starter/src/main/java/org/apache/seatunnel/edge/agent/starter/wal/sqlite/SqliteSourcePositionStore.java:89

        try (PreparedStatement statement =
                connection.prepareStatement(SourcePositionSqlStatements.SELECT_BY_SOURCE)) {
            statement.setString(1, sourceId);
            try (ResultSet resultSet = statement.executeQuery()) {
                while (resultSet.next()) {
                    EdgeSourcePosition position = toPosition(resultSet);
                    positions.put(position.getPartition(), position);
                }
            }
        }
        return positions;
    }

    @Override
    public void save(EdgeSourcePosition position) throws Exception {
        String sourceId = position.getSourceId();
        String partition = position.getPartition();
        if (sourceId == null || sourceId.trim().isEmpty()) {
            throw new IllegalArgumentException("source position sourceId must not be blank");
        }
        if (partition == null || partition.trim().isEmpty()) {
            throw new IllegalArgumentException("source position partition must not be blank");
        }
        long updatedAt =
                position.getUpdatedAt() > 0L ? position.getUpdatedAt() : System.currentTimeMillis();
        try (PreparedStatement statement =
                connection.prepareStatement(SourcePositionSqlStatements.UPSERT)) {
            statement.setString(1, sourceId);
            statement.setString(2, partition);
            statement.setLong(3, position.getOffset());
            statement.setBytes(4, MetadataSerde.serialize(position.getMetadata()));
            statement.setLong(5, updatedAt);
            statement.executeUpdate();
        }
    }

    @Override

View on GitHub (pinned to cf67b549a7)