Fix test. (#1896)

(cherry picked from commit ed2be1d)

Author: Technoboy-

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpExchange.java

@@ -34,10 +34,20 @@ public interface AmqpExchange {
      *
      */
     enum Type{
-        Direct,
-        Fanout,
-        Topic,
-        Headers;
+        Direct("direct"),
+        Fanout("fanout"),
+        Topic("topic"),
+        Headers("headers"),
+        ConsistentHash("x-consistent-hash"),
+        ModulusHash("x-modulus-hash"),
+        LocalRandom("x-local-random"),
+        Random("x-random");
+
+        private final String type;
+
+        Type(String type) {
+            this.type = type;
+        }
 
         public static Type value(String type) {
             if (type == null || type.length() == 0) {
@@ -53,11 +63,24 @@ public static Type value(String type) {
                     return Topic;
                 case "headers":
                     return Headers;
+                case "x-consistent-hash":
+                    return ConsistentHash;
+                case "x-modulus-hash":
+                    return ModulusHash;
+                case "x-local-random":
+                    return LocalRandom;
+                case "x-random":
+                    return Random;
                 default:
                     return null;
             }
         }
 
+        @Override
+        public String toString() {
+            return type;
+        }
+
     }
 
     /**

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/ExchangeMessageRouter.java

@@ -301,6 +301,8 @@ public static ExchangeMessageRouter getInstance(PersistentExchange exchange, Exe
             case Direct -> new DirectExchangeMessageRouter(exchange, routeExecutor);
             case Topic -> new TopicExchangeMessageRouter(exchange, routeExecutor);
             case Headers -> new HeadersExchangeMessageRouter(exchange, routeExecutor);
+            default -> throw new AoPServiceRuntimeException.NotSupportedOperationException(
+                    "Exchange router is not supported for type " + exchange.getType() + ".");
         };
     }
 

amqp-impl/src/test/java/io/streamnative/pulsar/handlers/amqp/test/ExchangeMessageRouterTest.java

@@ -0,0 +1,51 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.pulsar.handlers.amqp.test;
+
+import io.streamnative.pulsar.handlers.amqp.AmqpExchange;
+import io.streamnative.pulsar.handlers.amqp.ExchangeMessageRouter;
+import io.streamnative.pulsar.handlers.amqp.common.exception.AoPServiceRuntimeException.NotSupportedOperationException;
+import io.streamnative.pulsar.handlers.amqp.impl.PersistentExchange;
+import java.util.concurrent.ExecutorService;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/**
+ * Unit tests for exchange router creation.
+ */
+public class ExchangeMessageRouterTest {
+
+    @DataProvider(name = "unsupportedExchangeTypes")
+    public Object[][] unsupportedExchangeTypes() {
+        return new Object[][]{
+                {AmqpExchange.Type.ConsistentHash},
+                {AmqpExchange.Type.ModulusHash},
+                {AmqpExchange.Type.LocalRandom},
+                {AmqpExchange.Type.Random}
+        };
+    }
+
+    @Test(dataProvider = "unsupportedExchangeTypes")
+    public void shouldThrowWhenExchangeTypeHasNoRouter(AmqpExchange.Type exchangeType) {
+        PersistentExchange exchange = Mockito.mock(PersistentExchange.class);
+        ExecutorService routeExecutor = Mockito.mock(ExecutorService.class);
+        Mockito.when(exchange.getType()).thenReturn(exchangeType);
+
+        NotSupportedOperationException exception = Assert.expectThrows(NotSupportedOperationException.class,
+                () -> ExchangeMessageRouter.getInstance(exchange, routeExecutor));
+        Assert.assertTrue(exception.getMessage().contains(exchangeType.toString()));
+    }
+}

tests/src/test/java/io/streamnative/pulsar/handlers/amqp/rabbitmq/functional/ExchangeDeclareTest.java

@@ -118,8 +118,8 @@ public void exchangeDeclaredWithEnumerationEquivalentOnRecoverableConnection()
 
     private void doTestExchangeDeclaredWithEnumerationEquivalent(Channel channel)
             throws IOException, InterruptedException {
-        assertEquals("There are 4 standard exchange types",
-                4, BuiltinExchangeType.values().length);
+        assertEquals("There are 8 standard exchange types",
+                8, BuiltinExchangeType.values().length);
         for (BuiltinExchangeType exchangeType : BuiltinExchangeType.values()) {
             channel.exchangeDeclare(NAME, exchangeType);
             verifyEquivalent(NAME, exchangeType.getType(), false, false, null);