{"record":{"id":"64ffd5fc15031137","repo":"apache/druid","slug":"emit-called-unexpectedly-before-service-start","errorCode":null,"errorMessage":"Emit called unexpectedly before service start","messagePattern":"Emit called unexpectedly before service start","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"extensions-contrib/ambari-metrics-emitter/src/main/java/org/apache/druid/emitter/ambari/metrics/AmbariMetricsEmitter.java","lineNumber":111,"sourceCode":"          loadTruststore(config.getTrustStorePath(), config.getTrustStoreType(), config.getTrustStorePassword());\n        }\n        exec.scheduleAtFixedRate(\n            new ConsumerRunnable(),\n            config.getFlushPeriod(),\n            config.getFlushPeriod(),\n            TimeUnit.MILLISECONDS\n        );\n        started.set(true);\n      }\n    }\n  }\n\n\n  @Override\n  public void emit(Event event)\n  {\n    if (!started.get()) {\n      throw new ISE(\"Emit called unexpectedly before service start\");\n    }\n    if (event instanceof ServiceMetricEvent) {\n      final TimelineMetric timelineEvent = timelineMetricConverter.druidEventToTimelineMetric((ServiceMetricEvent) event);\n      if (timelineEvent == null) {\n        return;\n      }\n      try {\n        final boolean isSuccessful = eventsQueue.offer(\n            timelineEvent,\n            config.getEmitWaitTime(),\n            TimeUnit.MILLISECONDS\n        );\n        if (!isSuccessful) {\n          if (countLostEvents.getAndIncrement() % 1000 == 0) {\n            log.error(\n                \"Lost total of [%s] events because of emitter queue is full. Please increase the capacity or/and the consumer frequency\",\n                countLostEvents.get()\n            );","sourceCodeStart":93,"sourceCodeEnd":129,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-contrib/ambari-metrics-emitter/src/main/java/org/apache/druid/emitter/ambari/metrics/AmbariMetricsEmitter.java#L93-L129","documentation":"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.","triggerScenarios":"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.","commonSituations":"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).","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."],"exampleFix":"// before: merge whatever bytes come back from sub-agents\nbyte[] merged = filters.get(0);\nfor (byte[] f : filters) {\n  merged = BloomKFilter.mergeBloomFilterBytes(merged, f); // throws when params differ\n}\n\n// after: check header compatibility first\nBloomKFilter first = BloomKFilter.deserialize(ByteBuffer.wrap(filters.get(0)), 0);\nfor (byte[] f : filters) {\n  BloomKFilter other = BloomKFilter.deserialize(ByteBuffer.wrap(f), 0);\n  if (other.getNumHashFunctions() != first.getNumHashFunctions() || other.getBitSet().length != first.getBitSet().length) {\n    throw new IllegalStateException(\"bloom filter parameters differ; rebuild with same numBits/numHashFunctions\");\n  }\n}","handlingStrategy":"validation","validationCode":"boolean mergeable(byte[] a, byte[] b) {\n  for (int i = 0; i < BloomKFilter.START_OF_SERIALIZED_LONGS; i++) {\n    if (a[i] != b[i]) return false;\n  }\n  return a.length == b.length;\n}","typeGuard":"boolean isMergeable(byte[] bf1, byte[] bf2) {\n  return bf1 != null && bf2 != null\n      && bf1.length == bf2.length\n      && Arrays.equals(bf1, 0, BloomKFilter.START_OF_SERIALIZED_LONGS, bf2, 0, BloomKFilter.START_OF_SERIALIZED_LONGS);\n}","tryCatchPattern":"try {\n  merged = BloomKFilter.mergeBloomFilterBytes(bf1, bf2);\n} catch (IllegalArgumentException e) {\n  throw new IllegalStateException(\"bloom filters not built with same numBits/numHashFunctions: \" + e.getMessage(), e);\n}","preventionTips":["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."],"tags":["bloom-filter","argument-validation","aggregation","druid"],"backgroundTag":"invalid-argument-value","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}