apache/druid · error · IllegalStateException

No sink for identifier

Error message

No sink for identifier: %s

What it means

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.

Solutions

1. Avoid dropping sinks concurrently with persist; serialize handoff and persist operations.
2. Upgrade Druid if hitting a known race between drop and persist in your version.
3. Restart the task and re-ingest; inspect for custom code mutating the sinks map.

Example fix

// before
// concurrent appenderator.drop(identifier) while persistAll runs

// after
// acquire the appenderator lock / run persist and drop on the same executor
synchronized (appenderator) { appenderator.persistAll(committer); }

// before persist, ensure no concurrent drops
if (dropsInProgress.get() > 0) { deferPersist(); return; }

try { appenderator.persistAll(committer) } catch (IllegalStateException e) { if (e.getMessage().startsWith("No sink for identifier")) { logAndRetryAfterDropsComplete(e); } else throw e; }

When it happens

Trigger: Concurrent drop/abandon of a sink while persistAll iterates the sinks map, or external mutation of the map leaving null entries.

Common situations: Race between segment handoff (dropping sinks) and a persist triggered by add or push on the same appenderator; custom subclass misuse.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• Attempt to add row to swapped-out sink for segment
• bitmapEncoding[ ] does not match indexSpec.bitmap[ ]
• Cannot copy [%,d] bytes, only have [%,d] buffered
• Cannot delete [%,d] bytes, only have [%,d] buffered
• Cleaned partition map

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/07dc459bee4d3f4e. Report an issue: GitHub.

Appendix: source

Thrown at server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java:652

    }
  }

  @Override
  public ListenableFuture<Object> persistAll(@Nullable final Committer committer)
  {
    throwPersistErrorIfExists();
    final Map<String, Integer> currentHydrants = new HashMap<>();
    final List<Pair<FireHydrant, SegmentIdWithShardSpec>> indexesToPersist = new ArrayList<>();
    int numPersistedRows = 0;
    long bytesPersisted = 0L;
    MutableLong totalHydrantsCount = new MutableLong();
    MutableLong totalHydrantsPersisted = new MutableLong();
    final long totalSinks = sinks.size();
    for (Map.Entry<SegmentIdWithShardSpec, Sink> entry : sinks.entrySet()) {
      final SegmentIdWithShardSpec identifier = entry.getKey();
      final Sink sink = entry.getValue();
      if (sink == null) {
        throw new ISE("No sink for identifier: %s", identifier);
      }
      final List<FireHydrant> hydrants = Lists.newArrayList(sink);
      totalHydrantsCount.add(hydrants.size());
      currentHydrants.put(identifier.toString(), hydrants.size());
      numPersistedRows += sink.getNumRowsInMemory();
      bytesPersisted += sink.getBytesInMemory();

      final int limit = sink.isWritable() ? hydrants.size() - 1 : hydrants.size();

      // gather hydrants that have not been persisted:
      for (FireHydrant hydrant : hydrants.subList(0, limit)) {
        if (!hydrant.hasSwapped()) {
          log.debug("Hydrant[%s] hasn't persisted yet, persisting. Segment[%s]", hydrant, identifier);
          indexesToPersist.add(Pair.of(hydrant, identifier));
          totalHydrantsPersisted.add(1);
        }
      }

View on GitHub (pinned to 9b90983fd2)