{"record":{"id":"56674ffa8fa411ee","repo":"apache/druid","slug":"sink-must-not-be-null-for-identifier-when-persisti","errorCode":null,"errorMessage":"Sink must not be null for identifier when persisting hydrant[%s]","messagePattern":"Sink must not be null for identifier when persisting hydrant\\[(.+?)\\]","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java","lineNumber":1189,"sourceCode":"      throw new ISE(\n          \"Segment[%s] hydrant[%s] already swapped. This cannot happen.\",\n          identifier,\n          indexToPersist\n      );\n    }\n\n    log.debug(\"Segment[%s], persisting Hydrant[%s]\", identifier, indexToPersist);\n\n    try {\n      final long startTime = System.nanoTime();\n      int numRows = indexToPersist.getIndex().numRows();\n\n      // since the sink may have been persisted before it may have lost its\n      // hydrant count, we remember that value in the sinks' metadata, so we have\n      // to pull it from there....\n      SinkMetadata sm = sinksMetadata.get(identifier);\n      if (sm == null) {\n        throw new ISE(\"Sink must not be null for identifier when persisting hydrant[%s]\", identifier);\n      }\n      final File persistDir = createPersistDirIfNeeded(identifier);\n      indexMerger.persist(\n          indexToPersist.getIndex(),\n          identifier.getInterval(),\n          new File(persistDir, String.valueOf(sm.getNumHydrants())),\n          tuningConfig.getIndexSpecForIntermediatePersists(),\n          tuningConfig.getSegmentWriteOutMediumFactory()\n      );\n      sm.setPersistedFileDir(persistDir);\n\n      log.info(\n          \"Persisted in-memory data for segment[%s] spill[%s] to disk in [%,d] ms (%,d rows).\",\n          indexToPersist.getSegmentId(),\n          sm.getNumHydrants(),\n          (System.nanoTime() - startTime) / 1000000,\n          numRows\n      );","sourceCodeStart":1171,"sourceCodeEnd":1207,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java#L1171-L1207","documentation":"When persisting a hydrant, BatchAppenderator looks up the SinkMetadata for the identifier to learn the hydrant count carried in sink metadata. A null entry means persist was requested for an identifier that this appenderator does not currently track as an active sink, so it cannot determine the persist directory numbering — this ISE is thrown.","triggerScenarios":"Calling appenderator.persist(identifier...) (or the flush path reaching it) for an identifier that was never added, or that was already dropped/cleared via drop/ clearing of sinksMetadata — e.g. persist racing with a drop, or persisting after the batch job's finishTransaction/cleanup.","commonSituations":"Custom callers driving Appenderator APIs directly and persisting after shutdown/drop; identifiers from a previous job incarnation; race between persist and cleanup in orchestration code.","solutions":["Only call persist for identifiers returned from getCurrentSinks()/actively added segments in this appenderator instance","Ensure persist calls complete before drop()/close() is invoked — serialize persist and cleanup operations","Restart the batch job if state is already inconsistent; sinksMetadata cannot be resurrected after drop","Verify no two BatchAppenderatorDriver instances share one appenderator"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"if (appenderator.getCurrentSinks().stream()\n    .noneMatch(sink -> sink.getId().equals(identifier))) {\n  throw new IllegalStateException(\"Identifier not active: \" + identifier);\n}","typeGuard":"boolean isActiveSink(Appenderator a, SegmentIdWithShardSpec id) {\n  return a.getCurrentSinks().stream().anyMatch(s -> s.getId().equals(id));\n}","tryCatchPattern":"try {\n  appenderator.persist(identifier);\n} catch (ISE e) {\n  if (e.getMessage().contains(\"Sink must not be null\")) {\n    log.warn(\"Persist raced with drop/cleanup for %s\", identifier);\n  }\n  throw e;\n}","preventionTips":["Serialize persist calls with drop()/close() — never overlap them","Only persist identifiers obtained from getCurrentSinks() in this appenderator instance","Do not share one Appenderator between multiple drivers"],"tags":["appenderator","lifecycle","invariant"],"backgroundTag":"entity-not-found","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"}