You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "kfaraz (via GitHub)" <gi...@apache.org> on 2025/07/15 10:25:56 UTC
[PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
kfaraz opened a new pull request, #18255:
URL: https://github.com/apache/druid/pull/18255
Follow up to #18249 [discussion](https://github.com/apache/druid/pull/18249#discussion_r2206525247)
### 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
- Reduce the usage of `StubServiceEmitter.getEvents()`. Use the inbuilt `verifyValue` methods instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "Akshat-Jain (via GitHub)" <gi...@apache.org>.
Akshat-Jain closed pull request #18255: Minor clean up of LatchableEmitter / StubServiceEmitter
URL: https://github.com/apache/druid/pull/18255
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209172050
##########
server/src/test/java/org/apache/druid/server/ClientQuerySegmentWalkerTest.java:
##########
@@ -987,10 +986,11 @@ public void testMetricsWithMaxSubqueryRowsEnabled()
ImmutableList.of(new Object[]{Intervals.ETERNITY.getStartMillis(), 3L})
);
- List<Event> events = emitter.getEvents();
+ List<StubServiceEmitter.ServiceMetricEventSnapshot> events =
+ emitter.getMetricEvents(ClientQuerySegmentWalker.ROWS_COUNT_METRIC);
- for (Event event : events) {
- EventMap map = event.toMap();
+ for (StubServiceEmitter.ServiceMetricEventSnapshot event : events) {
Review Comment:
Yeah, I am not sure why this class `ServiceMetricEventSnapshot` was added either.
Let me see what I can do to clean up this part.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209760430
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kgyrtkirk (via GitHub)" <gi...@apache.org>.
kgyrtkirk commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209744437
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
Review Comment:
why the need to put it on a different thread? it doesn't seem like its a lot of computation...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18255:
URL: https://github.com/apache/druid/pull/18255#issuecomment-3073063953
@kgyrtkirk , I have tried to clean up the `LatchableEmitter` since it seemed very 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kgyrtkirk (via GitHub)" <gi...@apache.org>.
kgyrtkirk commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2207711845
##########
processing/src/test/java/org/apache/druid/java/util/metrics/StubServiceEmitter.java:
##########
@@ -74,14 +74,20 @@ public List<Event> getEvents()
return new ArrayList<>(events);
}
+ public int getNumEmittedEvents()
+ {
+ return events.size();
+ }
+
/**
- * Gets all the metric events emitted since the previous {@link #flush()}.
+ * Gets all the metric events emitted for the given metric name since the previous {@link #flush()}.
*
- * @return Map from metric name to list of events emitted for that metric.
+ * @return List of events emitted for the given metric.
*/
- public Map<String, Queue<ServiceMetricEventSnapshot>> getMetricEvents()
+ public List<ServiceMetricEventSnapshot> getMetricEvents(String metricName)
Review Comment:
this is exactly what I should have done in the previous PR....but rushed to get it running....
##########
server/src/test/java/org/apache/druid/server/ClientQuerySegmentWalkerTest.java:
##########
@@ -987,10 +986,11 @@ public void testMetricsWithMaxSubqueryRowsEnabled()
ImmutableList.of(new Object[]{Intervals.ETERNITY.getStartMillis(), 3L})
);
- List<Event> events = emitter.getEvents();
+ List<StubServiceEmitter.ServiceMetricEventSnapshot> events =
+ emitter.getMetricEvents(ClientQuerySegmentWalker.ROWS_COUNT_METRIC);
- for (Event event : events) {
- EventMap map = event.toMap();
+ for (StubServiceEmitter.ServiceMetricEventSnapshot event : events) {
Review Comment:
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` -
##########
services/src/test/java/org/apache/druid/server/AsyncQueryForwardingServletTest.java:
##########
@@ -683,7 +683,8 @@ protected void doService(
}
catch (NullPointerException ignored) {
}
- Assert.assertEquals("query/time", stubServiceEmitter.getEvents().get(0).toMap().get("metric"));
+ // Assert.assertEquals("query/time", stubServiceEmitter.getEvents().get(0).toMap().get("metric"));
Review Comment:
nit: commented old code
##########
server/src/test/java/org/apache/druid/server/ClientQuerySegmentWalkerTest.java:
##########
@@ -1038,23 +1038,26 @@ public void testMetricsWithMaxSubqueryBytesEnabled()
ImmutableList.of(new Object[]{Intervals.ETERNITY.getStartMillis(), 3L})
);
- List<Event> events = emitter.getEvents();
+ List<StubServiceEmitter.ServiceMetricEventSnapshot> events
+ = emitter.getMetricEvents(ClientQuerySegmentWalker.ROWS_COUNT_METRIC);
Review Comment:
nit: could be inlined into the `for`
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -60,7 +60,13 @@ public class LatchableEmitter extends StubServiceEmitter
*/
private final ScheduledExecutorService conditionEvaluateExecutor;
private final Set<WaitCondition> waitConditions = new HashSet<>();
- private final ReentrantReadWriteLock eventReadWriteLock = new ReentrantReadWriteLock(true);
+
+ private final ReentrantLock eventProcessingLock = new ReentrantLock();
+
+ /**
+ * Lists of events that have already been processed by {@link #evaluateWaitConditions(Event)}.
+ */
+ private final List<Event> processedEvents = new ArrayList<>();
Review Comment:
no `clear` is called on this field - even during `flush`
could that cause problems?
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
Review Comment:
this is partially unrelated; but I think this should be a serious error - (possibly the predicate is broken?)...repack into `RuntimeException` if necessary?
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -107,9 +93,9 @@ public void close()
public void waitForEvent(Predicate<Event> condition, long timeoutMillis)
{
final WaitCondition waitCondition = new WaitCondition(condition);
+ registerWaitCondition(waitCondition);
waitConditions.add(waitCondition);
Review Comment:
nit: `register` may already have added it to the `waitConditions`
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
}
finally {
- eventReadWriteLock.readLock().unlock();
+ processedEvents.add(event);
+ eventProcessingLock.unlock();
+ }
+ }
+
+ /**
+ * Evaluates the given new condition for all past events and then adds it to
+ * {@link #waitConditions}.
+ */
+ private void registerWaitCondition(WaitCondition condition)
+ {
+ eventProcessingLock.lock();
+ try {
+ for (Event event : processedEvents) {
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
+ }
+ }
+ waitConditions.add(condition);
Review Comment:
nit: we don't necessarily need to add if the condition is already satisfied
##########
server/src/test/java/org/apache/druid/server/metrics/ServiceStatusMonitorTest.java:
##########
@@ -47,36 +49,27 @@ public void setUp()
@Test
public void testDefaultHeartbeatReported()
{
- final StubServiceEmitter emitter = new StubServiceEmitter("service", "host");
monitor.doMonitor(emitter);
- Assert.assertEquals(1, emitter.getEvents().size());
- Assert.assertEquals(HEARTBEAT_METRIC_KEY, emitter.getEvents().get(0).toMap().get("metric"));
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("value"));
+ Assert.assertEquals(1, emitter.getNumEmittedEvents());
+ emitter.verifyValue(HEARTBEAT_METRIC_KEY, 1);
}
@Test
public void testLeaderTag()
{
heartbeatTags.put("leader", 1);
- final StubServiceEmitter emitter = new StubServiceEmitter("service", "host");
monitor.doMonitor(emitter);
- Assert.assertEquals(1, emitter.getEvents().size());
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("leader"));
- Assert.assertEquals(HEARTBEAT_METRIC_KEY, emitter.getEvents().get(0).toMap().get("metric"));
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("value"));
+ Assert.assertEquals(1, emitter.getNumEmittedEvents());
+ emitter.verifyValue(HEARTBEAT_METRIC_KEY, Map.of("leader", 1), 1);
}
@Test
public void testMoreThanOneTag()
{
heartbeatTags.put("leader", 1);
heartbeatTags.put("taskRunner", "http");
- final StubServiceEmitter emitter = new StubServiceEmitter("service", "host");
monitor.doMonitor(emitter);
- Assert.assertEquals(1, emitter.getEvents().size());
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("leader"));
- Assert.assertEquals("http", emitter.getEvents().get(0).toMap().get("taskRunner"));
- Assert.assertEquals(HEARTBEAT_METRIC_KEY, emitter.getEvents().get(0).toMap().get("metric"));
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("value"));
+ Assert.assertEquals(1, emitter.getNumEmittedEvents());
+ emitter.verifyValue(HEARTBEAT_METRIC_KEY, Map.of("leader", 1, "taskRunner", "http"), 1);
Review Comment:
pretty cool how concise this become! :)
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
}
finally {
- eventReadWriteLock.readLock().unlock();
+ processedEvents.add(event);
+ eventProcessingLock.unlock();
+ }
+ }
+
+ /**
+ * Evaluates the given new condition for all past events and then adds it to
+ * {@link #waitConditions}.
+ */
+ private void registerWaitCondition(WaitCondition condition)
+ {
+ eventProcessingLock.lock();
+ try {
+ for (Event event : processedEvents) {
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
+ }
Review Comment:
nit: shouldn't this be `condition#evaluate` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2207924618
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
}
finally {
- eventReadWriteLock.readLock().unlock();
+ processedEvents.add(event);
+ eventProcessingLock.unlock();
+ }
+ }
+
+ /**
+ * Evaluates the given new condition for all past events and then adds it to
+ * {@link #waitConditions}.
+ */
+ private void registerWaitCondition(WaitCondition condition)
+ {
+ eventProcessingLock.lock();
+ try {
+ for (Event event : processedEvents) {
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
+ }
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2207912132
##########
server/src/test/java/org/apache/druid/server/metrics/ServiceStatusMonitorTest.java:
##########
@@ -47,36 +49,27 @@ public void setUp()
@Test
public void testDefaultHeartbeatReported()
{
- final StubServiceEmitter emitter = new StubServiceEmitter("service", "host");
monitor.doMonitor(emitter);
- Assert.assertEquals(1, emitter.getEvents().size());
- Assert.assertEquals(HEARTBEAT_METRIC_KEY, emitter.getEvents().get(0).toMap().get("metric"));
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("value"));
+ Assert.assertEquals(1, emitter.getNumEmittedEvents());
+ emitter.verifyValue(HEARTBEAT_METRIC_KEY, 1);
}
@Test
public void testLeaderTag()
{
heartbeatTags.put("leader", 1);
- final StubServiceEmitter emitter = new StubServiceEmitter("service", "host");
monitor.doMonitor(emitter);
- Assert.assertEquals(1, emitter.getEvents().size());
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("leader"));
- Assert.assertEquals(HEARTBEAT_METRIC_KEY, emitter.getEvents().get(0).toMap().get("metric"));
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("value"));
+ Assert.assertEquals(1, emitter.getNumEmittedEvents());
+ emitter.verifyValue(HEARTBEAT_METRIC_KEY, Map.of("leader", 1), 1);
}
@Test
public void testMoreThanOneTag()
{
heartbeatTags.put("leader", 1);
heartbeatTags.put("taskRunner", "http");
- final StubServiceEmitter emitter = new StubServiceEmitter("service", "host");
monitor.doMonitor(emitter);
- Assert.assertEquals(1, emitter.getEvents().size());
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("leader"));
- Assert.assertEquals("http", emitter.getEvents().get(0).toMap().get("taskRunner"));
- Assert.assertEquals(HEARTBEAT_METRIC_KEY, emitter.getEvents().get(0).toMap().get("metric"));
- Assert.assertEquals(1, emitter.getEvents().get(0).toMap().get("value"));
+ Assert.assertEquals(1, emitter.getNumEmittedEvents());
+ emitter.verifyValue(HEARTBEAT_METRIC_KEY, Map.of("leader", 1, "taskRunner", "http"), 1);
Review Comment:
🙂
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209783858
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
Review Comment:
Fair enough, I will throw a runtime exception here. Hopefully, it helps us catch issues early.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18255:
URL: https://github.com/apache/druid/pull/18255#issuecomment-3076763924
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "Akshat-Jain (via GitHub)" <gi...@apache.org>.
Akshat-Jain closed pull request #18255: Minor clean up of LatchableEmitter / StubServiceEmitter
URL: https://github.com/apache/druid/pull/18255
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209748696
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
Review Comment:
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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2207913351
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
}
finally {
- eventReadWriteLock.readLock().unlock();
+ processedEvents.add(event);
+ eventProcessingLock.unlock();
+ }
+ }
+
+ /**
+ * Evaluates the given new condition for all past events and then adds it to
+ * {@link #waitConditions}.
+ */
+ private void registerWaitCondition(WaitCondition condition)
+ {
+ eventProcessingLock.lock();
+ try {
+ for (Event event : processedEvents) {
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
+ }
Review Comment:
Oops, let me fix this up!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2207914162
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
}
finally {
- eventReadWriteLock.readLock().unlock();
+ processedEvents.add(event);
+ eventProcessingLock.unlock();
+ }
+ }
+
+ /**
+ * Evaluates the given new condition for all past events and then adds it to
+ * {@link #waitConditions}.
+ */
+ private void registerWaitCondition(WaitCondition condition)
+ {
+ eventProcessingLock.lock();
+ try {
+ for (Event event : processedEvents) {
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
+ }
+ }
+ waitConditions.add(condition);
Review Comment:
Yeah, that's true, let me check this again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209181298
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209760430
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
Review Comment:
I am still not sure if we can throw an Exception here 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "Akshat-Jain (via GitHub)" <gi...@apache.org>.
Akshat-Jain closed pull request #18255: Minor clean up of LatchableEmitter / StubServiceEmitter
URL: https://github.com/apache/druid/pull/18255
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2207926748
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -60,7 +60,13 @@ public class LatchableEmitter extends StubServiceEmitter
*/
private final ScheduledExecutorService conditionEvaluateExecutor;
private final Set<WaitCondition> waitConditions = new HashSet<>();
- private final ReentrantReadWriteLock eventReadWriteLock = new ReentrantReadWriteLock(true);
+
+ private final ReentrantLock eventProcessingLock = new ReentrantLock();
+
+ /**
+ * Lists of events that have already been processed by {@link #evaluateWaitConditions(Event)}.
+ */
+ private final List<Event> processedEvents = new ArrayList<>();
Review Comment:
Thanks for catching this, yeah, let me clean this up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2207911214
##########
services/src/test/java/org/apache/druid/server/AsyncQueryForwardingServletTest.java:
##########
@@ -683,7 +683,8 @@ protected void doService(
}
catch (NullPointerException ignored) {
}
- Assert.assertEquals("query/time", stubServiceEmitter.getEvents().get(0).toMap().get("metric"));
+ // Assert.assertEquals("query/time", stubServiceEmitter.getEvents().get(0).toMap().get("metric"));
Review Comment:
Thanks for catching this!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2207926748
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -60,7 +60,13 @@ public class LatchableEmitter extends StubServiceEmitter
*/
private final ScheduledExecutorService conditionEvaluateExecutor;
private final Set<WaitCondition> waitConditions = new HashSet<>();
- private final ReentrantReadWriteLock eventReadWriteLock = new ReentrantReadWriteLock(true);
+
+ private final ReentrantLock eventProcessingLock = new ReentrantLock();
+
+ /**
+ * Lists of events that have already been processed by {@link #evaluateWaitConditions(Event)}.
+ */
+ private final List<Event> processedEvents = new ArrayList<>();
Review Comment:
Thanks for catching this, let me clean this up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz merged PR #18255:
URL: https://github.com/apache/druid/pull/18255
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209211750
##########
server/src/test/java/org/apache/druid/server/ClientQuerySegmentWalkerTest.java:
##########
@@ -987,10 +986,11 @@ public void testMetricsWithMaxSubqueryRowsEnabled()
ImmutableList.of(new Object[]{Intervals.ETERNITY.getStartMillis(), 3L})
);
- List<Event> events = emitter.getEvents();
+ List<StubServiceEmitter.ServiceMetricEventSnapshot> events =
+ emitter.getMetricEvents(ClientQuerySegmentWalker.ROWS_COUNT_METRIC);
- for (Event event : events) {
- EventMap map = event.toMap();
+ for (StubServiceEmitter.ServiceMetricEventSnapshot event : events) {
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209229691
##########
server/src/test/java/org/apache/druid/server/ClientQuerySegmentWalkerTest.java:
##########
@@ -987,10 +986,11 @@ public void testMetricsWithMaxSubqueryRowsEnabled()
ImmutableList.of(new Object[]{Intervals.ETERNITY.getStartMillis(), 3L})
);
- List<Event> events = emitter.getEvents();
+ List<StubServiceEmitter.ServiceMetricEventSnapshot> events =
+ emitter.getMetricEvents(ClientQuerySegmentWalker.ROWS_COUNT_METRIC);
- for (Event event : events) {
- EventMap map = event.toMap();
+ for (StubServiceEmitter.ServiceMetricEventSnapshot event : events) {
Review Comment:
Left a comment on the original PR https://github.com/apache/druid/pull/17170#discussion_r2209227633
I have added a `ServiceMetricEvent.copy()` for now as you suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Minor clean up of LatchableEmitter / StubServiceEmitter (druid)
Posted by "kgyrtkirk (via GitHub)" <gi...@apache.org>.
kgyrtkirk commented on code in PR #18255:
URL: https://github.com/apache/druid/pull/18255#discussion_r2209780962
##########
server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java:
##########
@@ -158,48 +144,64 @@ public void waitForEventAggregate(
);
}
- private void triggerConditionEvaluations()
+ private void triggerConditionEvaluations(Event event)
{
if (conditionEvaluateExecutor == null) {
throw new ISE("Cannot evaluate conditions as the 'conditionEvaluateExecutor' is null.");
} else {
- conditionEvaluateExecutor.submit(this::evaluateWaitConditions);
+ conditionEvaluateExecutor.submit(() -> evaluateWaitConditions(event));
}
}
/**
* Evaluates wait conditions. This method must be invoked on the
* {@link #conditionEvaluateExecutor} so that it does not block {@link #emit(Event)}.
*/
- private void evaluateWaitConditions()
+ private void evaluateWaitConditions(Event event)
{
- eventReadWriteLock.readLock().lock();
+ eventProcessingLock.lock();
try {
// Create a copy of the conditions for thread-safety
final List<WaitCondition> conditionsToEvaluate = List.copyOf(waitConditions);
if (conditionsToEvaluate.isEmpty()) {
return;
}
- List<Event> events = getEvents();
for (WaitCondition condition : conditionsToEvaluate) {
- final int currentNumberOfEvents = events.size();
-
- // Do not use an iterator over the list to avoid concurrent modification exceptions
- // Evaluate new events against this condition
- for (int i = condition.processedUntil; i < currentNumberOfEvents; ++i) {
- if (condition.predicate.test(events.get(i))) {
- condition.countDownLatch.countDown();
- }
+ if (condition.predicate.test(event)) {
+ condition.countDownLatch.countDown();
}
- condition.processedUntil = currentNumberOfEvents;
}
}
catch (Exception e) {
log.error(e, "Error while evaluating wait conditions");
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org