apache/druid · error · IllegalArgumentException

Expected dataSource[ ] but was asked to insert row for…

Error message

Expected dataSource[%s] but was asked to insert row for dataSource[%s]?!

What it means

StreamAppenderator.add validates that the identifier of the pending segment being written matches the dataSource of the appenderator's DataSchema. A mismatch means rows are being routed to an appenderator configured for a different datasource, so the insert is rejected with an IAE.

Solutions

1. Ensure the task/IO config datasource matches the dataSchema datasource exactly (case-sensitive).
2. Create one Appenderator per datasource instead of sharing the instance.
3. Fix custom code to build identifiers with the same dataSource as the appenderator's schema.

Example fix

// before
Appenderator appenderator = getAppenderatorFor("other-ds");
appenderator.add(identifierForDataSource("my-ds"), row, ...);

// after
Appenderator appenderator = getAppenderatorFor("my-ds");
appenderator.add(identifierForDataSource("my-ds"), row, ...);

if (!identifier.getDataSource().equals(schema.getDataSource())) {
  throw new IllegalArgumentException("datasource mismatch: " + identifier.getDataSource());
}

try { appenderator.add(id, row, supplier, false) } catch (IllegalArgumentException e) { if (e.getMessage().contains("Expected dataSource")) { reinitAppenderatorForDatasource(id.getDataSource()); } else throw e; }

When it happens

Trigger: Calling StreamAppenderator.add with a SegmentIdWithShardSpec whose getDataSource() differs from schema.getDataSource() — e.g. misconfigured firehose/tuning where events are attributed to another datasource, or reusing an appenderator instance across datasources.

Common situations: Custom ingestion code reusing an appenderator for multiple datasources; task spec datasource renamed but segment allocation from a stale coordinator assignment; Kafka index task with mismatched datasource in IO config vs dataSchema.

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

• argument should be an identifier expression. Use array()…
• Attempt to add row to swapped-out sink for segment
• Bloom filter aggregators are query-time only
• Can't find pushedSegments for segment
• Can't find segmentsForSequence for sequence

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/c8b647e7194f0e8c. Report an issue: GitHub.

Appendix: source

Thrown at server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java:315

  private void throwPersistErrorIfExists()
  {
    if (persistError != null) {
      throw new RE(persistError, "Error while persisting");
    }
  }

  @Override
  public AppenderatorAddResult add(
      final SegmentIdWithShardSpec identifier,
      final InputRow row,
      @Nullable final Supplier<Committer> committerSupplier,
      final boolean allowIncrementalPersists
  ) throws SegmentNotWritableException
  {
    throwPersistErrorIfExists();

    if (!identifier.getDataSource().equals(schema.getDataSource())) {
      throw new IAE(
          "Expected dataSource[%s] but was asked to insert row for dataSource[%s]?!",
          schema.getDataSource(),
          identifier.getDataSource()
      );
    }

    final Sink sink = getOrCreateSink(identifier);
    metrics.reportMessageMaxTimestamp(row.getTimestampFromEpoch());
    final int sinkRowsInMemoryBeforeAdd = sink.getNumRowsInMemory();
    final int sinkRowsInMemoryAfterAdd;
    final long bytesInMemoryBeforeAdd = sink.getBytesInMemory();
    final long bytesInMemoryAfterAdd;
    final IncrementalIndexAddResult addResult;

    addResult = sink.add(row);
    sinkRowsInMemoryAfterAdd = addResult.getRowCount();
    bytesInMemoryAfterAdd = addResult.getBytesInMemory();

View on GitHub (pinned to 9b90983fd2)