Minor clean up of LatchableEmitter / StubServiceEmitter by kfaraz · Pull Request #18255 · apache/druid

kfaraz · 2025-07-15T10:25:53Z

Changes

Maintain a List of processed events in LatchableEmitter.
This is an improvement over the current flow where a copy of events is created upon receiving every new event.
When a new condition is registered, evaluate all past events upfront, then add it to the set of wait conditions
Evaluate each new event as it is received

Other changes

Hide the internal queue implementation of StubServiceEmitter from both tests and the sub-class LatchableEmitter
Reduce the usage of StubServiceEmitter.getEvents(). Use the inbuilt verifyValue methods instead.

kfaraz · 2025-07-15T10:27:15Z

@kgyrtkirk , I have tried to clean up the LatchableEmitter since it seemed wasteful to create a new list every time.

I have also removed the processedUntil flag to simplify the flow, while retaining readability.
The LatchableEmitter is now agnostic of how StubServiceEmitter stores events.

Please let me know if this makes sense.

kgyrtkirk

Thank you @kfaraz, this cleaned up some stuff pretty nicely!
left a few notes here and there - I think only the comment on processedEvents matters

kgyrtkirk · 2025-07-15T14:59:18Z


-    for (Event event : events) {
-      EventMap map = event.toMap();
+    for (StubServiceEmitter.ServiceMetricEventSnapshot event : events) {


nit: right now there is event.getUserDims() and event.getMetricEvent().getUserDims() ; I have no clear understanding why this ServiceMetricEventSnapshot is necessary ; but if the userDims or some other things might be clobbered externally; it would be more straight to just add a copy constructor to ServiceMetricEvent -

Yeah, I am not sure why this class ServiceMetricEventSnapshot was added either.
Let me see what I can do to clean up this part.

Took a look at this. Apparently the userDims can change if the builder is reused for emitting another event.
But this is problematic anyway.

If the same event is to be emitted using (one or more) emitters, it is possible that before we even get to emit the event (as emission is always async), the dimensions of the event have changed.

I wonder if we should do this in a separate PR though as it is addressing a specific race condition. Let me know what you think.

Left a comment on the original PR #17170 (comment)
I have added a ServiceMetricEvent.copy() for now as you suggested.

kgyrtkirk · 2025-07-15T15:07:25Z

+        if (condition.predicate.test(event)) {
+          condition.countDownLatch.countDown();
+        }


nit: shouldn't this be condition#evaluate ?

Oops, let me fix this up!

I checked this. It seems correct. The condition is evaluated (using condition.predicate.test()) against all the events and the latch is counted down if the condition is satisfied at any point.
I think we can probably break the loop once the condition is satisfied.

kgyrtkirk · 2025-07-15T15:10:45Z

+          condition.countDownLatch.countDown();
+        }
+      }
+      waitConditions.add(condition);


nit: we don't necessarily need to add if the condition is already satisfied

Yeah, that's true, let me check this again.

kgyrtkirk · 2025-07-15T15:20:35Z

      }
    }
    catch (Exception e) {
      log.error(e, "Error while evaluating wait conditions");


this is partially unrelated; but I think this should be a serious error - (possibly the predicate is broken?)...repack into RuntimeException if necessary?

That would be good but this is running on a separate thread. So the thrown exception will not cause any real failure except for the countdown latch eventually timing out.

I could use futures instead of countdown latches, but the latch gives a cleaner flow.

Let me know if you have any suggestion to handle this.

why the need to put it on a different thread? it doesn't seem like its a lot of computation...

Hmm, yeah, you are right. The original design needed the other thread because some conditions had to be evaluated from the start. But now we evaluate only one event at a time, so I guess we can do it synchronously after all.

Thanks for the suggestion!

I am still not sure if throwing an exception here will help though.
The contract of Emitter.emit() says that it shouldn't throw an exception and only log warnings/errors.
An exception thrown here would still not fail the test directly as the emit() call might be happening on some service thread that just swallows up the exception.

I'm afraid we can't tell future bugs to not happen - so that apidoc is hard to live up to ; it may suggest to try to avoid that ; but the system invoking this api should fully prepared that something might go wrong even in this method

Fair enough, I will throw a runtime exception here. Hopefully, it helps us catch issues early.

kfaraz · 2025-07-16T05:06:28Z

@kgyrtkirk , thanks a lot for the prompt review and for the suggestions!
I have incorporated your feedback. Please let me know if the changes look good.

…mitter

Follow up to apache#18249 Changes: - Maintain a List of processed events in `LatchableEmitter`. This is an improvement over the current flow where a copy of events is created upon receiving every new event. - When a new condition is registered, evaluate all past events upfront, then add it to the set of wait conditions - Evaluate each new event as it is received Other changes: - Hide the internal queue implementation of `StubServiceEmitter` from tests and sub-classes - Reduce the usage of `StubServiceEmitter.getEvents()`. Use the inbuilt `verifyValue` methods instead.

Minor clean up of LatchableEmitter / StubServiceEmitter

9733ff5

kfaraz requested a review from kgyrtkirk July 15, 2025 10:26

kgyrtkirk approved these changes Jul 15, 2025

View reviewed changes

Akshat-Jain closed this Jul 15, 2025

Akshat-Jain reopened this Jul 15, 2025

github-actions Bot added Kubernetes Area - Ingestion labels Jul 15, 2025

Remove ServiceMetricEventSnapshot

a6329b3

Merge branch 'master' of github.com:apache/druid into clean_up_stub_e…

58ec921

…mitter

Akshat-Jain closed this Jul 16, 2025

Akshat-Jain reopened this Jul 16, 2025

github-actions Bot added the Area - Metrics/Event Emitting label Jul 16, 2025

kfaraz added 3 commits July 16, 2025 13:24

Add locking around close, flush in LatchableEmitter

c59b48e

Evaluate conditions synchronously

f5e8054

throw exception if condition eval fails

9056935

Akshat-Jain closed this Jul 16, 2025

Akshat-Jain reopened this Jul 16, 2025

kfaraz merged commit 7aec41b into apache:master Jul 16, 2025
76 checks passed

kfaraz deleted the clean_up_stub_emitter branch July 16, 2025 14:13

kfaraz mentioned this pull request Jul 17, 2025

Make ServiceMetricEvent immutable, do not allow null dimension values #18272

Merged

10 tasks

cecemei added this to the 35.0.0 milestone Oct 21, 2025

Conversation

kfaraz commented Jul 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changes

Other changes

Uh oh!

kfaraz commented Jul 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kgyrtkirk left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kfaraz Jul 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

kfaraz commented Jul 16, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

kfaraz commented Jul 15, 2025 •

edited

Loading

kfaraz commented Jul 15, 2025 •

edited

Loading

kfaraz Jul 16, 2025 •

edited

Loading