Followup multiple eventlistener plugin changes (Fix tests).

ScrapCodes · ScrapCodes · commit d3c9ef0f3d1d · 2025-03-19T17:50:29.000+05:30
A previous PR
added the needed changes to support Multiple event listener.
However it missed some changes in tests, query runners etc.
So here we fix the follow up, i.e. fix tests etc..
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveDistributedQueries.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveDistributedQueries.java
@@ -24,7 +24,7 @@
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.Test;
 
-import java.util.Optional;
+import java.util.List;
 import java.util.Set;
 
 import static com.facebook.presto.SystemSessionProperties.CTE_MATERIALIZATION_STRATEGY;
@@ -60,10 +60,10 @@ protected boolean supportsNotNullColumns()
     public void close()
             throws Exception
     {
-        Optional<EventListener> eventListener = getQueryRunner().getEventListener();
-        assertTrue(eventListener.isPresent());
-        assertTrue(eventListener.get() instanceof TestingHiveEventListener, eventListener.get().getClass().getName());
-        Set<QueryId> runningQueryIds = ((TestingHiveEventListener) eventListener.get()).getRunningQueries();
+        List<EventListener> eventListener = getQueryRunner().getEventListeners();
+        assertTrue(!eventListener.isEmpty());
+        assertTrue(eventListener.get(0) instanceof TestingHiveEventListener, eventListener.get(0).getClass().getName());
+        Set<QueryId> runningQueryIds = ((TestingHiveEventListener) eventListener.get(0)).getRunningQueries();
 
         if (!runningQueryIds.isEmpty()) {
             // Await query events to propagate and finish
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveSplitScheduling.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveSplitScheduling.java
@@ -21,10 +21,12 @@
 import com.google.common.collect.ImmutableMap;
 import org.testng.annotations.Test;
 
+import java.util.List;
 import java.util.Optional;
 
 import static com.facebook.presto.hive.HiveSessionProperties.DYNAMIC_SPLIT_SIZES_ENABLED;
 import static io.airlift.tpch.TpchTable.ORDERS;
+import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
 @Test(singleThreaded = true)
@@ -72,10 +74,10 @@ public void testDynamicSplits()
 
     private TestHiveEventListenerPlugin.TestingHiveEventListener getEventListener()
     {
-        Optional<EventListener> eventListener = getQueryRunner().getEventListener();
-        assertTrue(eventListener.isPresent());
-        assertTrue(eventListener.get() instanceof TestHiveEventListenerPlugin.TestingHiveEventListener, eventListener.get().getClass().getName());
-        return (TestHiveEventListenerPlugin.TestingHiveEventListener) eventListener.get();
+        List<EventListener> eventListener = getQueryRunner().getEventListeners();
+        assertFalse(eventListener.isEmpty());
+        assertTrue(eventListener.get(0) instanceof TestHiveEventListenerPlugin.TestingHiveEventListener, eventListener.get(0).getClass().getName());
+        return (TestHiveEventListenerPlugin.TestingHiveEventListener) eventListener.get(0);
     }
 
     private Session dynamicSplitsSession()
diff --git a/presto-main/src/main/java/com/facebook/presto/server/testing/TestingPrestoServer.java b/presto-main/src/main/java/com/facebook/presto/server/testing/TestingPrestoServer.java
@@ -628,10 +628,10 @@ public StatsCalculator getStatsCalculator()
         return statsCalculator;
     }
 
-    public Optional<EventListener> getEventListener()
+    public List<EventListener> getEventListeners()
     {
         checkState(coordinator, "not a coordinator");
-        return eventListenerManager.getEventListener();
+        return eventListenerManager.getEventListeners();
     }
 
     public TestingAccessControlManager getAccessControl()
diff --git a/presto-main/src/main/java/com/facebook/presto/testing/LocalQueryRunner.java b/presto-main/src/main/java/com/facebook/presto/testing/LocalQueryRunner.java
@@ -713,9 +713,9 @@ public StatsCalculator getStatsCalculator()
     }
 
     @Override
-    public Optional<EventListener> getEventListener()
+    public List<EventListener> getEventListeners()
     {
-        return Optional.empty();
+        return ImmutableList.of();
     }
 
     public CostCalculator getCostCalculator()
diff --git a/presto-main/src/main/java/com/facebook/presto/testing/QueryRunner.java b/presto-main/src/main/java/com/facebook/presto/testing/QueryRunner.java
@@ -60,7 +60,16 @@ public interface QueryRunner
 
     StatsCalculator getStatsCalculator();
 
-    Optional<EventListener> getEventListener();
+    @Deprecated
+    default Optional<EventListener> getEventListener()
+    {
+        if (getEventListeners().isEmpty()) {
+            return Optional.empty();
+        }
+        return Optional.of(getEventListeners().get(0));
+    }
+
+    List<EventListener> getEventListeners();
 
     TestingAccessControlManager getAccessControl();
 
diff --git a/presto-main/src/main/java/com/facebook/presto/testing/TestingEventListenerManager.java b/presto-main/src/main/java/com/facebook/presto/testing/TestingEventListenerManager.java
@@ -22,16 +22,17 @@
 import com.facebook.presto.spi.eventlistener.QueryProgressEvent;
 import com.facebook.presto.spi.eventlistener.QueryUpdatedEvent;
 import com.facebook.presto.spi.eventlistener.SplitCompletedEvent;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.inject.Inject;
 
-import java.util.Optional;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class TestingEventListenerManager
         extends EventListenerManager
 {
-    private final AtomicReference<Optional<EventListener>> configuredEventListener = new AtomicReference<>(Optional.empty());
+    private final AtomicReference<List<EventListener>> configuredEventListeners = new AtomicReference<>(ImmutableList.of());
 
     @Inject
     public TestingEventListenerManager(EventListenerConfig config)
@@ -42,51 +43,46 @@ public TestingEventListenerManager(EventListenerConfig config)
     @Override
     public void addEventListenerFactory(EventListenerFactory eventListenerFactory)
     {
-        configuredEventListener.set(Optional.of(eventListenerFactory.create(ImmutableMap.of())));
+        configuredEventListeners.set(ImmutableList.of(eventListenerFactory.create(ImmutableMap.of())));
     }
 
     @Override
     public void queryCompleted(QueryCompletedEvent queryCompletedEvent)
     {
-        if (configuredEventListener.get().isPresent()) {
-            configuredEventListener.get().get().queryCompleted(queryCompletedEvent);
-        }
+        configuredEventListeners.get()
+                .forEach(eventListener -> eventListener.queryCompleted(queryCompletedEvent));
     }
 
     @Override
     public void queryCreated(QueryCreatedEvent queryCreatedEvent)
     {
-        if (configuredEventListener.get().isPresent()) {
-            configuredEventListener.get().get().queryCreated(queryCreatedEvent);
-        }
+        configuredEventListeners.get()
+                .forEach(eventListener -> eventListener.queryCreated(queryCreatedEvent));
     }
 
     @Override
     public void queryUpdated(QueryUpdatedEvent queryUpdatedEvent)
     {
-        if (configuredEventListener.get().isPresent()) {
-            configuredEventListener.get().get().queryUpdated(queryUpdatedEvent);
-        }
+        configuredEventListeners.get()
+                .forEach(eventListener -> eventListener.queryUpdated(queryUpdatedEvent));
     }
 
     @Override
     public void publishQueryProgress(QueryProgressEvent queryProgressEvent)
     {
-        if (configuredEventListener.get().isPresent()) {
-            configuredEventListener.get().get().publishQueryProgress(queryProgressEvent);
-        }
+        configuredEventListeners.get()
+                .forEach(eventListener -> eventListener.publishQueryProgress(queryProgressEvent));
     }
 
     @Override
     public void splitCompleted(SplitCompletedEvent splitCompletedEvent)
     {
-        if (configuredEventListener.get().isPresent()) {
-            configuredEventListener.get().get().splitCompleted(splitCompletedEvent);
-        }
+        configuredEventListeners.get()
+                .forEach(eventListener -> eventListener.splitCompleted(splitCompletedEvent));
     }
 
-    public Optional<EventListener> getEventListener()
+    public List<EventListener> getEventListeners()
     {
-        return configuredEventListener.get();
+        return configuredEventListeners.get();
     }
 }
diff --git a/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/ContainerQueryRunner.java b/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/ContainerQueryRunner.java
@@ -45,7 +45,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.logging.Logger;
@@ -223,7 +222,7 @@ public StatsCalculator getStatsCalculator()
     }
 
     @Override
-    public Optional<EventListener> getEventListener()
+    public List<EventListener> getEventListeners()
     {
         throw new UnsupportedOperationException();
     }
diff --git a/presto-spark-base/src/test/java/com/facebook/presto/spark/PrestoSparkQueryRunner.java b/presto-spark-base/src/test/java/com/facebook/presto/spark/PrestoSparkQueryRunner.java
@@ -498,9 +498,9 @@ public StatsCalculator getStatsCalculator()
     }
 
     @Override
-    public Optional<EventListener> getEventListener()
+    public List<EventListener> getEventListeners()
     {
-        return Optional.empty();
+        return ImmutableList.of();
     }
 
     @Override
diff --git a/presto-tests/src/main/java/com/facebook/presto/tests/DistributedQueryRunner.java b/presto-tests/src/main/java/com/facebook/presto/tests/DistributedQueryRunner.java
@@ -607,10 +607,10 @@ public StatsCalculator getStatsCalculator()
     }
 
     @Override
-    public Optional<EventListener> getEventListener()
+    public List<EventListener> getEventListeners()
     {
         checkState(coordinators.size() == 1, "Expected a single coordinator");
-        return coordinators.get(0).getEventListener();
+        return coordinators.get(0).getEventListeners();
     }
 
     @Override
diff --git a/presto-tests/src/main/java/com/facebook/presto/tests/StandaloneQueryRunner.java b/presto-tests/src/main/java/com/facebook/presto/tests/StandaloneQueryRunner.java
@@ -42,7 +42,6 @@
 
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
@@ -178,9 +177,9 @@ public StatsCalculator getStatsCalculator()
     }
 
     @Override
-    public Optional<EventListener> getEventListener()
+    public List<EventListener> getEventListeners()
     {
-        return server.getEventListener();
+        return server.getEventListeners();
     }
 
     @Override
diff --git a/presto-thrift-connector/src/test/java/com/facebook/presto/connector/thrift/integration/ThriftQueryRunner.java b/presto-thrift-connector/src/test/java/com/facebook/presto/connector/thrift/integration/ThriftQueryRunner.java
@@ -50,7 +50,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.locks.Lock;
 
 import static com.facebook.airlift.testing.Closeables.closeQuietly;
@@ -244,9 +243,9 @@ public StatsCalculator getStatsCalculator()
         }
 
         @Override
-        public Optional<EventListener> getEventListener()
+        public List<EventListener> getEventListeners()
         {
-            return source.getEventListener();
+            return source.getEventListeners();
         }
 
         @Override

Original file line number	Diff line number	Diff line change
`@@ -628,10 +628,10 @@ public StatsCalculator getStatsCalculator()`
`628`	`628`	`return statsCalculator;`
`629`	`629`	`}`
`630`	`630`
`631`		`- public Optional<EventListener> getEventListener()`
	`631`	`+ public List<EventListener> getEventListeners()`
`632`	`632`	`{`
`633`	`633`	`checkState(coordinator, "not a coordinator");`
`634`		`- return eventListenerManager.getEventListener();`
	`634`	`+ return eventListenerManager.getEventListeners();`
`635`	`635`	`}`
`636`	`636`
`637`	`637`	`public TestingAccessControlManager getAccessControl()`
Original file line number	Diff line number	Diff line change
`@@ -713,9 +713,9 @@ public StatsCalculator getStatsCalculator()`
`713`	`713`	`}`
`714`	`714`
`715`	`715`	`@Override`
`716`		`- public Optional<EventListener> getEventListener()`
	`716`	`+ public List<EventListener> getEventListeners()`
`717`	`717`	`{`
`718`		`- return Optional.empty();`
	`718`	`+ return ImmutableList.of();`
`719`	`719`	`}`
`720`	`720`
`721`	`721`	`public CostCalculator getCostCalculator()`
Original file line number	Diff line number	Diff line change
`@@ -22,16 +22,17 @@`
`22`	`22`	`import com.facebook.presto.spi.eventlistener.QueryProgressEvent;`
`23`	`23`	`import com.facebook.presto.spi.eventlistener.QueryUpdatedEvent;`
`24`	`24`	`import com.facebook.presto.spi.eventlistener.SplitCompletedEvent;`
	`25`	`+import com.google.common.collect.ImmutableList;`
`25`	`26`	`import com.google.common.collect.ImmutableMap;`
`26`	`27`	`import com.google.inject.Inject;`
`27`	`28`
`28`		`-import java.util.Optional;`
	`29`	`+import java.util.List;`
`29`	`30`	`import java.util.concurrent.atomic.AtomicReference;`
`30`	`31`
`31`	`32`	`public class TestingEventListenerManager`
`32`	`33`	`extends EventListenerManager`
`33`	`34`	`{`
`34`		`- private final AtomicReference<Optional<EventListener>> configuredEventListener = new AtomicReference<>(Optional.empty());`
	`35`	`+ private final AtomicReference<List<EventListener>> configuredEventListeners = new AtomicReference<>(ImmutableList.of());`
`35`	`36`
`36`	`37`	`@Inject`
`37`	`38`	`public TestingEventListenerManager(EventListenerConfig config)`
`@@ -42,51 +43,46 @@ public TestingEventListenerManager(EventListenerConfig config)`
`42`	`43`	`@Override`
`43`	`44`	`public void addEventListenerFactory(EventListenerFactory eventListenerFactory)`
`44`	`45`	`{`
`45`		`- configuredEventListener.set(Optional.of(eventListenerFactory.create(ImmutableMap.of())));`
	`46`	`+ configuredEventListeners.set(ImmutableList.of(eventListenerFactory.create(ImmutableMap.of())));`
`46`	`47`	`}`
`47`	`48`
`48`	`49`	`@Override`
`49`	`50`	`public void queryCompleted(QueryCompletedEvent queryCompletedEvent)`
`50`	`51`	`{`
`51`		`- if (configuredEventListener.get().isPresent()) {`
`52`		`- configuredEventListener.get().get().queryCompleted(queryCompletedEvent);`
`53`		`- }`
	`52`	`+ configuredEventListeners.get()`
	`53`	`+ .forEach(eventListener -> eventListener.queryCompleted(queryCompletedEvent));`
`54`	`54`	`}`
`55`	`55`
`56`	`56`	`@Override`
`57`	`57`	`public void queryCreated(QueryCreatedEvent queryCreatedEvent)`
`58`	`58`	`{`
`59`		`- if (configuredEventListener.get().isPresent()) {`
`60`		`- configuredEventListener.get().get().queryCreated(queryCreatedEvent);`
`61`		`- }`
	`59`	`+ configuredEventListeners.get()`
	`60`	`+ .forEach(eventListener -> eventListener.queryCreated(queryCreatedEvent));`
`62`	`61`	`}`
`63`	`62`
`64`	`63`	`@Override`
`65`	`64`	`public void queryUpdated(QueryUpdatedEvent queryUpdatedEvent)`
`66`	`65`	`{`
`67`		`- if (configuredEventListener.get().isPresent()) {`
`68`		`- configuredEventListener.get().get().queryUpdated(queryUpdatedEvent);`
`69`		`- }`
	`66`	`+ configuredEventListeners.get()`
	`67`	`+ .forEach(eventListener -> eventListener.queryUpdated(queryUpdatedEvent));`
`70`	`68`	`}`
`71`	`69`
`72`	`70`	`@Override`
`73`	`71`	`public void publishQueryProgress(QueryProgressEvent queryProgressEvent)`
`74`	`72`	`{`
`75`		`- if (configuredEventListener.get().isPresent()) {`
`76`		`- configuredEventListener.get().get().publishQueryProgress(queryProgressEvent);`
`77`		`- }`
	`73`	`+ configuredEventListeners.get()`
	`74`	`+ .forEach(eventListener -> eventListener.publishQueryProgress(queryProgressEvent));`
`78`	`75`	`}`
`79`	`76`
`80`	`77`	`@Override`
`81`	`78`	`public void splitCompleted(SplitCompletedEvent splitCompletedEvent)`
`82`	`79`	`{`
`83`		`- if (configuredEventListener.get().isPresent()) {`
`84`		`- configuredEventListener.get().get().splitCompleted(splitCompletedEvent);`
`85`		`- }`
	`80`	`+ configuredEventListeners.get()`
	`81`	`+ .forEach(eventListener -> eventListener.splitCompleted(splitCompletedEvent));`
`86`	`82`	`}`
`87`	`83`
`88`		`- public Optional<EventListener> getEventListener()`
	`84`	`+ public List<EventListener> getEventListeners()`
`89`	`85`	`{`
`90`		`- return configuredEventListener.get();`
	`86`	`+ return configuredEventListeners.get();`
`91`	`87`	`}`
`92`	`88`	`}`
Original file line number	Diff line number	Diff line change
`@@ -45,7 +45,6 @@`
`45`	`45`	`import java.util.ArrayList;`
`46`	`46`	`import java.util.List;`
`47`	`47`	`import java.util.Map;`
`48`		`-import java.util.Optional;`
`49`	`48`	`import java.util.concurrent.TimeUnit;`
`50`	`49`	`import java.util.concurrent.locks.Lock;`
`51`	`50`	`import java.util.logging.Logger;`
`@@ -223,7 +222,7 @@ public StatsCalculator getStatsCalculator()`
`223`	`222`	`}`
`224`	`223`
`225`	`224`	`@Override`
`226`		`- public Optional<EventListener> getEventListener()`
	`225`	`+ public List<EventListener> getEventListeners()`
`227`	`226`	`{`
`228`	`227`	`throw new UnsupportedOperationException();`
`229`	`228`	`}`
Original file line number	Diff line number	Diff line change
`@@ -498,9 +498,9 @@ public StatsCalculator getStatsCalculator()`
`498`	`498`	`}`
`499`	`499`
`500`	`500`	`@Override`
`501`		`- public Optional<EventListener> getEventListener()`
	`501`	`+ public List<EventListener> getEventListeners()`
`502`	`502`	`{`
`503`		`- return Optional.empty();`
	`503`	`+ return ImmutableList.of();`
`504`	`504`	`}`
`505`	`505`
`506`	`506`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -607,10 +607,10 @@ public StatsCalculator getStatsCalculator()`
`607`	`607`	`}`
`608`	`608`
`609`	`609`	`@Override`
`610`		`- public Optional<EventListener> getEventListener()`
	`610`	`+ public List<EventListener> getEventListeners()`
`611`	`611`	`{`
`612`	`612`	`checkState(coordinators.size() == 1, "Expected a single coordinator");`
`613`		`- return coordinators.get(0).getEventListener();`
	`613`	`+ return coordinators.get(0).getEventListeners();`
`614`	`614`	`}`
`615`	`615`
`616`	`616`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,6 @@`
`42`	`42`
`43`	`43`	`import java.util.List;`
`44`	`44`	`import java.util.Map;`
`45`		`-import java.util.Optional;`
`46`	`45`	`import java.util.Set;`
`47`	`46`	`import java.util.concurrent.locks.Lock;`
`48`	`47`	`import java.util.concurrent.locks.ReadWriteLock;`
`@@ -178,9 +177,9 @@ public StatsCalculator getStatsCalculator()`
`178`	`177`	`}`
`179`	`178`
`180`	`179`	`@Override`
`181`		`- public Optional<EventListener> getEventListener()`
	`180`	`+ public List<EventListener> getEventListeners()`
`182`	`181`	`{`
`183`		`- return server.getEventListener();`
	`182`	`+ return server.getEventListeners();`
`184`	`183`	`}`
`185`	`184`
`186`	`185`	`@Override`