apache/druid · error · IllegalStateException

Emit called unexpectedly before service start

Error message

Emit called unexpectedly before service start

What it means

BloomKFilter.mergeBloomFilterBytes merges two serialized bloom filters by bitwise-ORing their bitsets, which is only valid when both filters were built with the same number of hash functions and the same bitset size. Before merging, the header bytes (numHashFunctions/numBits fields, before START_OF_SERIALIZED_LONGS) are compared byte-for-byte, and any mismatch throws this IllegalArgumentException. This is a fail-fast guard against silently corrupting the merged filter.

Source

Thrown at extensions-contrib/ambari-metrics-emitter/src/main/java/org/apache/druid/emitter/ambari/metrics/AmbariMetricsEmitter.java:111

          loadTruststore(config.getTrustStorePath(), config.getTrustStoreType(), config.getTrustStorePassword());
        }
        exec.scheduleAtFixedRate(
            new ConsumerRunnable(),
            config.getFlushPeriod(),
            config.getFlushPeriod(),
            TimeUnit.MILLISECONDS
        );
        started.set(true);
      }
    }
  }


  @Override
  public void emit(Event event)
  {
    if (!started.get()) {
      throw new ISE("Emit called unexpectedly before service start");
    }
    if (event instanceof ServiceMetricEvent) {
      final TimelineMetric timelineEvent = timelineMetricConverter.druidEventToTimelineMetric((ServiceMetricEvent) event);
      if (timelineEvent == null) {
        return;
      }
      try {
        final boolean isSuccessful = eventsQueue.offer(
            timelineEvent,
            config.getEmitWaitTime(),
            TimeUnit.MILLISECONDS
        );
        if (!isSuccessful) {
          if (countLostEvents.getAndIncrement() % 1000 == 0) {
            log.error(
                "Lost total of [%s] events because of emitter queue is full. Please increase the capacity or/and the consumer frequency",
                countLostEvents.get()
            );

View on GitHub (pinned to 9b90983fd2)

Solutions

Ensure all filters being merged use identical bloomFilterNumBits and bloomFilterNumHashFunctions settings in every query/stage that produces them.
Before merging, deserialize both filters and compare getNumHashFunctions() and the bitset length; only merge filters that match.
If filters differ, rebuild them with canonical parameters instead of merging, or use multiple separate filters per parameter set.
Catch IllegalArgumentException from merge and surface a clear error identifying the mismatched parameters.

Example fix

// before: merge whatever bytes come back from sub-agents
byte[] merged = filters.get(0);
for (byte[] f : filters) {
  merged = BloomKFilter.mergeBloomFilterBytes(merged, f); // throws when params differ
}

// after: check header compatibility first
BloomKFilter first = BloomKFilter.deserialize(ByteBuffer.wrap(filters.get(0)), 0);
for (byte[] f : filters) {
  BloomKFilter other = BloomKFilter.deserialize(ByteBuffer.wrap(f), 0);
  if (other.getNumHashFunctions() != first.getNumHashFunctions() || other.getBitSet().length != first.getBitSet().length) {
    throw new IllegalStateException("bloom filter parameters differ; rebuild with same numBits/numHashFunctions");
  }
}

Defensive patterns

Strategy: validation

Validate before calling

boolean mergeable(byte[] a, byte[] b) {
  for (int i = 0; i < BloomKFilter.START_OF_SERIALIZED_LONGS; i++) {
    if (a[i] != b[i]) return false;
  }
  return a.length == b.length;
}

Type guard

boolean isMergeable(byte[] bf1, byte[] bf2) {
  return bf1 != null && bf2 != null
      && bf1.length == bf2.length
      && Arrays.equals(bf1, 0, BloomKFilter.START_OF_SERIALIZED_LONGS, bf2, 0, BloomKFilter.START_OF_SERIALIZED_LONGS);
}

Try / catch

try {
  merged = BloomKFilter.mergeBloomFilterBytes(bf1, bf2);
} catch (IllegalArgumentException e) {
  throw new IllegalStateException("bloom filters not built with same numBits/numHashFunctions: " + e.getMessage(), e);
}

Prevention

Pin bloomFilterNumBits and bloomFilterNumHashFunctions in a shared query config, not per-query.
Validate filter compatibility (header equality) before merging instead of relying on the throw.
Keep all Druid versions producing mergeable filters on the same BloomKFilter parameters.

When it happens

Trigger: Calling BloomKFilter.mergeBloomFilterBytes (directly or via the bloom filter aggregate/function on serialized filter bytes) with two filters whose serialized headers differ: different numHashFunctions or different numBits values in the first START_OF_SERIALIZED_LONGS bytes.

Common situations: Aggregating bloom filters produced with different `bloomFilterNumBits`/`bloomFilterNumHashFunctions` query parameters or different Druid versions/configs; merging filters built over columns of very different cardinality where sizes were tuned differently; mixing filters produced by external tools (e.g. Hive BloomKFilter with different parameters).

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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