fix test

streamnative · Jun 7, 2021 · 27c9fff · 27c9fff
1 parent 44012e0
commit 27c9fff
Show file tree

Hide file tree

Showing 7 changed files with 26 additions and 11 deletions.
diff --git a/.github/workflows/pr-test.yml b/.github/workflows/pr-test.yml
@@ -42,6 +42,9 @@ jobs:
     - name: Spotbugs check
       run: mvn spotbugs:check
 
+    - name: amqp-impl tests
+      run: mvn test -DfailIfNoTests=false -pl amqp-impl
+
     - name: test after build
       run: mvn test -DfailIfNoTests=false -pl tests
 

diff --git a/.../src/main/java/io/streamnative/pulsar/handlers/amqp/proxy/PulsarServiceLookupHandler.java b/.../src/main/java/io/streamnative/pulsar/handlers/amqp/proxy/PulsarServiceLookupHandler.java
@@ -27,6 +27,7 @@
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.metadata.api.MetadataCache;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
 import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
 
 /**
@@ -39,12 +40,12 @@ public class PulsarServiceLookupHandler implements LookupHandler {
 
     private PulsarClientImpl pulsarClient;
 
-    private MetadataCache<ServiceLookupData> serviceLookupDataCache;
+    private MetadataCache<LocalBrokerData> serviceLookupDataCache;
 
     public PulsarServiceLookupHandler(PulsarService pulsarService, PulsarClientImpl pulsarClient) {
         this.pulsarService = pulsarService;
         this.pulsarClient = pulsarClient;
-        this.serviceLookupDataCache = pulsarService.getLocalMetadataStore().getMetadataCache(ServiceLookupData.class);
+        this.serviceLookupDataCache = pulsarService.getLocalMetadataStore().getMetadataCache(LocalBrokerData.class);
     }
 
     @Override
@@ -73,7 +74,7 @@ public CompletableFuture<Pair<String, Integer>> findBroker(TopicName topicName,
                 }
             }
 
-            List<CompletableFuture<Optional<ServiceLookupData>>> futureList = new ArrayList<>();
+            List<CompletableFuture<Optional<LocalBrokerData>>> futureList = new ArrayList<>();
             for (String webService : matchWebUri) {
                 String path = LoadManager.LOADBALANCE_BROKERS_ROOT + "/" + webService;
                 futureList.add(serviceLookupDataCache.get(path));
@@ -109,8 +110,8 @@ public CompletableFuture<Pair<String, Integer>> findBroker(TopicName topicName,
                     return;
                 }
                 boolean match = false;
-                for (CompletableFuture<Optional<ServiceLookupData>> future : futureList) {
-                    Optional<ServiceLookupData> optionalServiceLookupData = future.join();
+                for (CompletableFuture<Optional<LocalBrokerData>> future : futureList) {
+                    Optional<LocalBrokerData> optionalServiceLookupData = future.join();
                     if (!optionalServiceLookupData.isPresent()) {
                         log.warn("Service lookup data is null.");
                         continue;

diff --git a/amqp-impl/src/main/resources/log4j2.xml b/amqp-impl/src/main/resources/log4j2.xml
@@ -36,7 +36,7 @@
         </Console>
     </Appenders>
     <Loggers>
-        <Root level="warn">
+        <Root level="info">
             <AppenderRef ref="Console" />
         </Root>
         <Logger name="org.eclipse.jetty" level="info"/>

diff --git a/amqp-impl/src/test/java/io/streamnative/pulsar/handlers/amqp/test/AmqpProtocolTestBase.java b/amqp-impl/src/test/java/io/streamnative/pulsar/handlers/amqp/test/AmqpProtocolTestBase.java
@@ -144,6 +144,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
         Mockito.when(pulsarService.getConfiguration()).thenReturn(serviceConfiguration);
         Mockito.when(pulsarService.getOrderedExecutor()).thenReturn(
                 OrderedExecutor.newBuilder().numThreads(8).name("pulsar-ordered").build());
+        Mockito.when(serviceConfiguration.getNumIOThreads()).thenReturn(2 * Runtime.getRuntime().availableProcessors());
     }
 
     private void mockBrokerService() {

diff --git a/tests/pom.xml b/tests/pom.xml
@@ -71,6 +71,12 @@
       <artifactId>qpid-test-utils</artifactId>
       <version>${qpid-protocol-plugin.version}</version>
       <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>ch.qos.logback</groupId>
+          <artifactId>logback-classic</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
 
     <dependency>

diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/amqp/AmqpProtocolHandlerTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/amqp/AmqpProtocolHandlerTestBase.java
@@ -18,6 +18,7 @@
 
 import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.MoreExecutors;
+import io.netty.channel.EventLoopGroup;
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.net.URI;
@@ -33,8 +34,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
-
-import io.netty.channel.EventLoopGroup;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.client.BookKeeper;
@@ -346,12 +345,17 @@ public CompletableFuture<ZooKeeper> create(String serverList, SessionType sessio
     private BookKeeperClientFactory mockBookKeeperClientFactory = new BookKeeperClientFactory() {
 
         @Override
-        public BookKeeper create(ServiceConfiguration conf, ZooKeeper zkClient, EventLoopGroup eventLoopGroup, Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass, Map<String, Object> ensemblePlacementPolicyProperties) throws IOException {
+        public BookKeeper create(ServiceConfiguration conf, ZooKeeper zkClient, EventLoopGroup eventLoopGroup,
+                                 Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
+                                 Map<String, Object> ensemblePlacementPolicyProperties) throws IOException {
             return mockBookKeeper;
         }
 
         @Override
-        public BookKeeper create(ServiceConfiguration conf, ZooKeeper zkClient, EventLoopGroup eventLoopGroup, Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass, Map<String, Object> ensemblePlacementPolicyProperties, StatsLogger statsLogger) throws IOException {
+        public BookKeeper create(ServiceConfiguration conf, ZooKeeper zkClient, EventLoopGroup eventLoopGroup,
+                                 Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
+                                 Map<String, Object> ensemblePlacementPolicyProperties, StatsLogger statsLogger)
+                throws IOException {
             return mockBookKeeper;
         }
 

diff --git a/tests/src/test/resources/log4j2.xml b/tests/src/test/resources/log4j2.xml
@@ -36,7 +36,7 @@
         </Console>
     </Appenders>
     <Loggers>
-        <Root level="warn">
+        <Root level="info">
             <AppenderRef ref="Console" />
         </Root>
         <Logger name="org.eclipse.jetty" level="info"/>