Followup multiple eventlistener plugin changes (Fix tests).

ScrapCodes · ScrapCodes · commit d84634425fbb · 2025-03-19T17:42:28.000+05:30
A previous PR #24456 added the needed changes to support Multiple event listener in tests, query runners and across the codebase.
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`