{"record":{"id":"c8b647e7194f0e8c","repo":"apache/druid","slug":"expected-datasource-s-but-was-asked-to-insert-ro","errorCode":null,"errorMessage":"Expected dataSource[%s] but was asked to insert row for dataSource[%s]?!","messagePattern":"Expected dataSource\\[(.+?)\\] but was asked to insert row for dataSource\\[(.+?)\\]\\?!","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java","lineNumber":315,"sourceCode":"  private void throwPersistErrorIfExists()\n  {\n    if (persistError != null) {\n      throw new RE(persistError, \"Error while persisting\");\n    }\n  }\n\n  @Override\n  public AppenderatorAddResult add(\n      final SegmentIdWithShardSpec identifier,\n      final InputRow row,\n      @Nullable final Supplier<Committer> committerSupplier,\n      final boolean allowIncrementalPersists\n  ) throws SegmentNotWritableException\n  {\n    throwPersistErrorIfExists();\n\n    if (!identifier.getDataSource().equals(schema.getDataSource())) {\n      throw new IAE(\n          \"Expected dataSource[%s] but was asked to insert row for dataSource[%s]?!\",\n          schema.getDataSource(),\n          identifier.getDataSource()\n      );\n    }\n\n    final Sink sink = getOrCreateSink(identifier);\n    metrics.reportMessageMaxTimestamp(row.getTimestampFromEpoch());\n    final int sinkRowsInMemoryBeforeAdd = sink.getNumRowsInMemory();\n    final int sinkRowsInMemoryAfterAdd;\n    final long bytesInMemoryBeforeAdd = sink.getBytesInMemory();\n    final long bytesInMemoryAfterAdd;\n    final IncrementalIndexAddResult addResult;\n\n    addResult = sink.add(row);\n    sinkRowsInMemoryAfterAdd = addResult.getRowCount();\n    bytesInMemoryAfterAdd = addResult.getBytesInMemory();\n","sourceCodeStart":297,"sourceCodeEnd":333,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java#L297-L333","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure the task/IO config datasource matches the dataSchema datasource exactly (case-sensitive).","Create one Appenderator per datasource instead of sharing the instance.","Fix custom code to build identifiers with the same dataSource as the appenderator's schema."],"exampleFix":"// before\nAppenderator appenderator = getAppenderatorFor(\"other-ds\");\nappenderator.add(identifierForDataSource(\"my-ds\"), row, ...);\n\n// after\nAppenderator appenderator = getAppenderatorFor(\"my-ds\");\nappenderator.add(identifierForDataSource(\"my-ds\"), row, ...);","handlingStrategy":"validation","validationCode":"if (!identifier.getDataSource().equals(schema.getDataSource())) {\n  throw new IllegalArgumentException(\"datasource mismatch: \" + identifier.getDataSource());\n}","typeGuard":null,"tryCatchPattern":"try { appenderator.add(id, row, supplier, false) } catch (IllegalArgumentException e) { if (e.getMessage().contains(\"Expected dataSource\")) { reinitAppenderatorForDatasource(id.getDataSource()); } else throw e; }","preventionTips":["Keep datasource name in one config place shared by dataSchema and IO config","Never share an Appenderator across datasources","Add startup assertion comparing identifier datasource to schema datasource"],"tags":["druid","ingestion","argument-validation"],"backgroundTag":"invalid-argument-value","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}