{"record":{"id":"f8e31dc6e93f0640","repo":"apache/druid","slug":"datasource-metadata-instance-does-not-match-requir","errorCode":null,"errorMessage":"Datasource metadata instance does not match required, found instance of [%s]","messagePattern":"Datasource metadata instance does not match required, found instance of \\[(.+?)\\]","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java","lineNumber":2156,"sourceCode":"    if (dataSourceMetadata == null) {\n      // Reset everything\n      boolean result = indexerMetadataStorageCoordinator.deleteDataSourceMetadata(supervisorId);\n      log.info(\"Reset supervisor[%s] for dataSource[%s] - dataSource metadata entry deleted? [%s]\", supervisorId, dataSource, result);\n      activelyReadingTaskGroups.values()\n                               .forEach(group -> killTasksInGroup(\n                                   group,\n                                   \"DataSourceMetadata is not found while reset\"\n                               ));\n      activelyReadingTaskGroups.clear();\n      partitionGroups.clear();\n      partitionOffsets.clear();\n      if (ioConfig.isBounded()) {\n        initializeBoundedPartitionGroups();\n        stateManager.maybeSetState(SupervisorStateManager.BasicState.RUNNING);\n      }\n    } else {\n      if (!checkSourceMetadataMatch(dataSourceMetadata)) {\n        throw new IAE(\n            \"Datasource metadata instance does not match required, found instance of [%s]\",\n            dataSourceMetadata.getClass()\n        );\n      }\n      log.info(\"Reset supervisor[%s] for dataSource[%s] with metadata[%s]\", supervisorId, dataSource, dataSourceMetadata);\n      // Reset only the partitions in dataSourceMetadata if it has not been reset yet\n      @SuppressWarnings(\"unchecked\")\n      final SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType> resetMetadata =\n          (SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType>) dataSourceMetadata;\n\n      if (resetMetadata.getSeekableStreamSequenceNumbers().getStream().equals(ioConfig.getStream())) {\n        // metadata can be null\n        final DataSourceMetadata metadata = indexerMetadataStorageCoordinator.retrieveDataSourceMetadata(supervisorId);\n        if (metadata != null && !checkSourceMetadataMatch(metadata)) {\n          throw new IAE(\n              \"Datasource metadata instance does not match required, found instance of [%s]\",\n              metadata.getClass()\n          );","sourceCodeStart":2138,"sourceCodeEnd":2174,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java#L2138-L2174","documentation":"The supervisor's reset endpoint (resetDataSourceMetadata) accepts a DataSourceMetadata describing what to reset. If the requested metadata is not the type the supervisor's IOConfig expects (checkSourceMetadataMatch fails), the reset cannot proceed and an IAE is thrown naming the received class.","triggerScenarios":"POSTing reset metadata whose class does not match the supervisor's source type — e.g. sending KafkaDataSourceMetadata to a Kinesis supervisor, or a generic DataSourceMetadata to a seekable-stream supervisor.","commonSituations":"Copy-pasted reset API calls between supervisors of different stream types; automation scripts resetting the wrong datasource; migrating ingestion from Kafka to Kinesis with old reset payloads.","solutions":["Send reset metadata of the class matching the supervisor's stream type (e.g. SeekableStreamDataSourceMetadata subclass for Kafka/Kinesis)","Verify the datasource name in the reset request targets the intended supervisor","Check the supervisor's ioConfig.stream/type before constructing reset metadata"],"exampleFix":"// before: wrong metadata type for a Kinesis supervisor\nclient.post('/druid/indexer/v1/supervisor/kinesis-svc/reset', new KafkaDataSourceMetadata(...));\n// after\nclient.post('/druid/indexer/v1/supervisor/kinesis-svc/reset',\n    new KinesisDataSourceMetadata(new SeekableStreamEndSequenceNumbers<>(stream, offsets)));","handlingStrategy":"validation","validationCode":"if (!metadata.getClass().equals(expectedMetadataClass)) {\n  throw new IllegalArgumentException(\"Reset metadata must be of type \" + expectedMetadataClass.getSimpleName());\n}","typeGuard":"boolean isMatchingMetadata(DataSourceMetadata m, Class<? extends DataSourceMetadata> required) {\n  return required.isInstance(m);\n}","tryCatchPattern":null,"preventionTips":["Match reset payload class to the supervisor's stream type (Kafka vs Kinesis)","Confirm the datasource name before issuing resets from scripts","Reuse the supervisor's spec ioConfig to build reset metadata"],"tags":["reset","type-mismatch","api"],"backgroundTag":"type-mismatch","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"}