apache/druid · error · IllegalStateException

Expected sink to be no longer writable before mergeAndPush…

Error message

Expected sink to be no longer writable before mergeAndPush for segment[%s].

What it means

Before merging and pushing a Sink, StreamAppenderator asserts the sink is no longer writable (finishWriting() was called) and all hydrants have been persisted. If the sink is still writable, the ISE is thrown because merging an actively-writable sink would produce inconsistent segment data.

Solutions

1. Always call finishWriting() (e.g. via appenderator.close() / persistAll) before push()
2. Ensure persist of all hydrants completes before mergeAndPush; check hasSwapped() on each hydrant
3. Inspect for concurrent writers still using the sink and stop them before push
4. If state is unrecoverable, restart the task to rebuild sinks from persisted segments

Example fix

// before
appenderator.push(identifiers, committer, null);
// after
for (SegmentIdWithShardSpec id : identifiers) {
  Sink sink = appenderator.getSink(id);
  if (sink != null && sink.isWritable()) {
    sink.finishWriting();
  }
}
appenderator.persistAll(committerSupplier.get()).get();
appenderator.push(identifiers, committer, null);

if (sink.isWritable()) {
  sink.finishWriting();
}
if (!sink.getHydrants().stream().allMatch(FireHydrant::hasSwapped)) {
  throw new IllegalStateException("Sink not fully persisted before push");
}

try {
  appenderator.push(ids, committer, null);
} catch (ISE e) {
  log.error("Sink state invalid before push: %s", e.getMessage());
  appenderator.persistAll(committer).get();
  throw e; // restart task to recover
}

When it happens

Trigger: Calling mergeAndPush() on a sink without calling finishWriting() first, or after a failed/partial persist left the sink in writable state; concurrent writes still active during push; a crashed persist left hydrants unswapped.

Common situations: Custom ingestion code or a task calling push() while ingestion is still adding rows; Kafka tasks killed mid-persist then retried against stale in-memory state; bugs in handoff sequencing.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• A batch appenderator was already created for this peon's…
• A realtime appenderator was already created for this peon's…
• Attempt to add row to swapped-out sink for segment
• authResult.getErrorMessage()
• Bloom filter aggregators are query-time only

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

Appendix: source

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

      final boolean useUniquePath
  )
  {
    // Bail out if this sink is null or otherwise not what we expect.
    //noinspection ObjectEquality
    if (sinks.get(identifier) != sink) {
      log.warn("Sink for segment[%s] no longer valid, bailing out of mergeAndPush.", identifier);
      return null;
    }

    // Use a descriptor file to indicate that pushing has completed.
    final File persistDir = computePersistDir(identifier);
    final File mergedTarget = new File(persistDir, "merged");
    final File descriptorFile = computeDescriptorFile(identifier);

    // Sanity checks
    for (FireHydrant hydrant : sink) {
      if (sink.isWritable()) {
        throw new ISE("Expected sink to be no longer writable before mergeAndPush for segment[%s].", identifier);
      }

      synchronized (hydrant) {
        if (!hydrant.hasSwapped()) {
          throw new ISE("Expected sink to be fully persisted before mergeAndPush for segment[%s].", identifier);
        }
      }
    }

    try {
      if (descriptorFile.exists()) {
        // Already pushed.

        if (useUniquePath) {
          // Don't reuse the descriptor, because the caller asked for a unique path. Leave the old one as-is, since
          // it might serve some unknown purpose.
          log.debug(
              "Segment[%s] already pushed, but we want a unique path, so will push again with a new path.",

View on GitHub (pinned to 9b90983fd2)