{"record":{"id":"07dc459bee4d3f4e","repo":"apache/druid","slug":"no-sink-for-identifier-s","errorCode":null,"errorMessage":"No sink for identifier: %s","messagePattern":"No sink for identifier: (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java","lineNumber":652,"sourceCode":"    }\n  }\n\n  @Override\n  public ListenableFuture<Object> persistAll(@Nullable final Committer committer)\n  {\n    throwPersistErrorIfExists();\n    final Map<String, Integer> currentHydrants = new HashMap<>();\n    final List<Pair<FireHydrant, SegmentIdWithShardSpec>> indexesToPersist = new ArrayList<>();\n    int numPersistedRows = 0;\n    long bytesPersisted = 0L;\n    MutableLong totalHydrantsCount = new MutableLong();\n    MutableLong totalHydrantsPersisted = new MutableLong();\n    final long totalSinks = sinks.size();\n    for (Map.Entry<SegmentIdWithShardSpec, Sink> entry : sinks.entrySet()) {\n      final SegmentIdWithShardSpec identifier = entry.getKey();\n      final Sink sink = entry.getValue();\n      if (sink == null) {\n        throw new ISE(\"No sink for identifier: %s\", identifier);\n      }\n      final List<FireHydrant> hydrants = Lists.newArrayList(sink);\n      totalHydrantsCount.add(hydrants.size());\n      currentHydrants.put(identifier.toString(), hydrants.size());\n      numPersistedRows += sink.getNumRowsInMemory();\n      bytesPersisted += sink.getBytesInMemory();\n\n      final int limit = sink.isWritable() ? hydrants.size() - 1 : hydrants.size();\n\n      // gather hydrants that have not been persisted:\n      for (FireHydrant hydrant : hydrants.subList(0, limit)) {\n        if (!hydrant.hasSwapped()) {\n          log.debug(\"Hydrant[%s] hasn't persisted yet, persisting. Segment[%s]\", hydrant, identifier);\n          indexesToPersist.add(Pair.of(hydrant, identifier));\n          totalHydrantsPersisted.add(1);\n        }\n      }\n","sourceCodeStart":634,"sourceCodeEnd":670,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java#L634-L670","documentation":"StreamAppenderator.persistAll iterates the sinks map and throws ISE('No sink for identifier: %s') if the map yields a null sink value — a defensive invariant check, since the map should never contain null values. It indicates corrupted internal state or concurrent modification of sinks during persist.","triggerScenarios":"Concurrent drop/abandon of a sink while persistAll iterates the sinks map, or external mutation of the map leaving null entries.","commonSituations":"Race between segment handoff (dropping sinks) and a persist triggered by add or push on the same appenderator; custom subclass misuse.","solutions":["Avoid dropping sinks concurrently with persist; serialize handoff and persist operations.","Upgrade Druid if hitting a known race between drop and persist in your version.","Restart the task and re-ingest; inspect for custom code mutating the sinks map."],"exampleFix":"// before\n// concurrent appenderator.drop(identifier) while persistAll runs\n\n// after\n// acquire the appenderator lock / run persist and drop on the same executor\nsynchronized (appenderator) { appenderator.persistAll(committer); }","handlingStrategy":"try-catch","validationCode":"// before persist, ensure no concurrent drops\nif (dropsInProgress.get() > 0) { deferPersist(); return; }","typeGuard":null,"tryCatchPattern":"try { appenderator.persistAll(committer) } catch (IllegalStateException e) { if (e.getMessage().startsWith(\"No sink for identifier\")) { logAndRetryAfterDropsComplete(e); } else throw e; }","preventionTips":["Serialize persist and drop operations on one executor","Upgrade to a Druid version with sink-drop race fixes","Avoid custom subclasses mutating the sinks map"],"tags":["druid","persistence","concurrency","invariant"],"backgroundTag":"internal-invariant-violation","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"}