{"record":{"id":"ff35de4aab830a01","repo":"apache/druid","slug":"unable-to-create-recordsupplier-s","errorCode":null,"errorMessage":"Unable to create RecordSupplier: %s","messagePattern":"Unable to create RecordSupplier: (.+?)","errorType":"exception","errorClass":"SamplerException","httpStatus":null,"severity":"error","filePath":"indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamSamplerSpec.java","lineNumber":77,"sourceCode":"    this.dataSchema = Preconditions.checkNotNull(ingestionSpec, \"[spec] is required\").getDataSchema();\n    this.ioConfig = Preconditions.checkNotNull(ingestionSpec.getIoConfig(), \"[spec.ioConfig] is required\");\n    this.tuningConfig = ingestionSpec.getTuningConfig();\n    this.samplerConfig = samplerConfig == null ? SamplerConfig.empty() : samplerConfig;\n    this.inputSourceSampler = inputSourceSampler;\n  }\n\n  @Override\n  public SamplerResponse sample()\n  {\n    final InputSource inputSource;\n    final InputFormat inputFormat;\n    RecordSupplier<PartitionIdType, SequenceOffsetType, RecordType> recordSupplier;\n\n    try {\n      recordSupplier = createRecordSupplier();\n    }\n    catch (Exception e) {\n      throw new SamplerException(e, \"Unable to create RecordSupplier: %s\", Throwables.getRootCause(e).getMessage());\n    }\n\n    inputSource = new RecordSupplierInputSource<>(\n        ioConfig.getStream(),\n        recordSupplier,\n        ioConfig.isUseEarliestSequenceNumber(),\n        samplerConfig.getTimeoutMs() <= 0 ? null : samplerConfig.getTimeoutMs()\n    );\n    inputFormat = Preconditions.checkNotNull(\n        ioConfig.getInputFormat(),\n        \"[spec.ioConfig.inputFormat] is required\"\n    );\n\n    return inputSourceSampler.sample(inputSource, inputFormat, dataSchema, samplerConfig);\n  }\n\n  protected abstract RecordSupplier<PartitionIdType, SequenceOffsetType, RecordType> createRecordSupplier();\n}","sourceCodeStart":59,"sourceCodeEnd":95,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamSamplerSpec.java#L59-L95","documentation":"SeekableStreamSamplerSpec.sample() creates the RecordSupplier (the Kafka/Kinesis consumer adapter) before sampling data. If createRecordSupplier() throws — bad consumer properties, missing config, incompatible dependency, provider/class loading failure — the exception is wrapped in a SamplerException with the root-cause message, prefixed 'Unable to create RecordSupplier'.","triggerScenarios":"Calling the sampling endpoint for a Kafka/Kinesis ingestion spec when createRecordSupplier() fails: missing bootstrap servers, invalid consumer config keys, wrong SASL/SSL settings, or the Kafka/Kinesis client libraries failing to initialize (missing extension jar, bad credentials provider).","commonSituations":"Kafka extension not loaded so consumer class/config is unavailable; typos in consumerProperties (e.g. bootstrap.servers missing); invalid AWS region/credentials for Kinesis; incompatible client library versions causing InstantiationException.","solutions":["Read the root-cause message after the colon and fix that specific config problem (e.g. add bootstrap.servers, fix SASL properties).","Confirm the druid-kafka-indexing-service / druid-kinesis-indexing-service extension is loaded on the router/middle manager.","Validate consumerProperties against the client library's documented config keys — remove or correct invalid entries.","For Kinesis, verify AWS region and credentials (config, env, IAM role) resolve before sampling."],"exampleFix":"// before: missing required consumer property\n\"ioConfig\": {\"topic\": \"events\", \"consumerProperties\": {}}\n// after: provide bootstrap servers\n\"ioConfig\": {\"topic\": \"events\", \"consumerProperties\": {\"bootstrap.servers\": \"broker1:9092,broker2:9092\"}}","handlingStrategy":"validation","validationCode":"// sanity-check required consumer properties before sampling\nMap<String,Object> props = spec.getIoConfig().getConsumerProperties();\nif (!props.containsKey(\"bootstrap.servers\")) { throw new ValidationException(\"bootstrap.servers required\"); }","typeGuard":null,"tryCatchPattern":"try { sampler.sample(); } catch (SamplerException e) { log.error(\"RecordSupplier init failed: {}\", Throwables.getRootCause(e).getMessage()); }","preventionTips":["Load the kafka/kinesis extensions on all ingest+router nodes","Validate consumerProperties keys against the client library docs","Verify AWS credentials/region for Kinesis before sampling"],"tags":["druid","kafka","kinesis","sampler","configuration"],"backgroundTag":"module-init-failed","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}