Avoid division by 0 when computing partition

Fixes #271

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/HashRoutingStrategy.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2021-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -33,10 +33,14 @@ class HashRoutingStrategy implements RoutingStrategy {
 
   @Override
   public List<String> route(Message message, Metadata metadata) {
-    String routingKey = routingKeyExtractor.apply(message);
-    int hashValue = hash.applyAsInt(routingKey);
     List<String> partitions = metadata.partitions();
-    return Collections.singletonList(
-        partitions.get(Integer.remainderUnsigned(hashValue, partitions.size())));
+    if (partitions.isEmpty()) {
+      return Collections.emptyList();
+    } else {
+      String routingKey = routingKeyExtractor.apply(message);
+      int hashValue = hash.applyAsInt(routingKey);
+      return Collections.singletonList(
+          partitions.get(Integer.remainderUnsigned(hashValue, partitions.size())));
+    }
   }
 }

src/main/java/com/rabbitmq/stream/impl/SuperStreamProducer.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2021-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -167,6 +167,9 @@ private DefaultSuperStreamMetadata(String superStream, StreamEnvironment environ
                   c -> c.partitions(superStream),
                   "Partition lookup for super stream '%s'",
                   superStream));
+      if (ps.isEmpty()) {
+        throw new IllegalArgumentException("Super stream '" + superStream + "' has no partition");
+      }
       this.partitions = new CopyOnWriteArrayList<>(ps);
     }
 

src/test/java/com/rabbitmq/stream/impl/HashRoutingStrategyTest.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -19,6 +19,7 @@
 import com.rabbitmq.stream.RoutingStrategy;
 import com.rabbitmq.stream.RoutingStrategy.Metadata;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -62,23 +63,43 @@ public String apply(Message message) {
         };
 
     Metadata metadata =
-        new Metadata() {
+        new MetadataAdapter() {
           List<String> partitions = Arrays.asList("invoices-01", "invoices-02", "invoices-03");
 
           @Override
           public List<String> partitions() {
             return partitions;
           }
-
-          @Override
-          public List<String> route(String routingKey) {
-            // not used here
-            return null;
-          }
         };
     for (String key : keys) {
       List<String> partitions = routingStrategy.route(null, metadata);
       assertThat(partitions).hasSize(1).element(0).isEqualTo(expectedRoutes.get(key));
     }
   }
+
+  @Test
+  void shouldReturnEmptyListIfNoPartition() {
+    RoutingStrategy routingStrategy = new HashRoutingStrategy(m -> "rk", HashUtils.MURMUR3);
+    Metadata metadata =
+        new MetadataAdapter() {
+          @Override
+          public List<String> partitions() {
+            return Collections.emptyList();
+          }
+        };
+    assertThat(routingStrategy.route(null, metadata)).isEmpty();
+  }
+
+  private static class MetadataAdapter implements Metadata {
+
+    @Override
+    public List<String> partitions() {
+      return null;
+    }
+
+    @Override
+    public List<String> route(String routingKey) {
+      return null;
+    }
+  }
 }

src/test/java/com/rabbitmq/stream/impl/SuperStreamProducerTest.java

@@ -331,4 +331,22 @@ void producerShouldNotPublishMessagesOnceClosed() throws Exception {
     assertThat(latchAssert(publishLatch)).completes(5);
     assertThat(confirmationCodes).hasSize(1).containsExactly(Constants.CODE_PRODUCER_CLOSED);
   }
+
+  @Test
+  void producerCreationShouldFailIfNoPartition() throws Exception {
+    declareSuperStreamTopology(connection, superStream, 0);
+    String producerName = "super-stream-application";
+    assertThatThrownBy(
+            () -> {
+              environment
+                  .producerBuilder()
+                  .name(producerName)
+                  .superStream(superStream)
+                  .routing(message -> message.getProperties().getMessageIdAsString())
+                  .producerBuilder()
+                  .build();
+            })
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("no partition");
+  }
 }