Limit segment size to 3 GB

Fixes #132

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/ByteCapacity.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 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").
@@ -16,12 +16,13 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.function.BiFunction;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 /** API to easily configure byte capacities. */
-public class ByteCapacity {
+public class ByteCapacity implements Comparable<ByteCapacity> {
 
   private static final String UNIT_MB = "mb";
   private static final int KILOBYTES_MULTIPLIER = 1000;
@@ -123,8 +124,30 @@ public static ByteCapacity from(String value) {
     }
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ByteCapacity that = (ByteCapacity) o;
+    return bytes == that.bytes;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(bytes);
+  }
+
   @Override
   public String toString() {
     return this.input;
   }
+
+  @Override
+  public int compareTo(ByteCapacity other) {
+    return Long.compare(this.bytes, other.bytes);
+  }
 }

src/main/java/com/rabbitmq/stream/StreamCreator.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 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").
@@ -18,6 +18,9 @@
 /** API to configure and create a stream. */
 public interface StreamCreator {
 
+  /** Segment size is limited to 3 GB. */
+  ByteCapacity MAX_SEGMENT_SIZE = ByteCapacity.from("3GB");
+
   /**
    * The name of the stream
    *
@@ -37,6 +40,8 @@ public interface StreamCreator {
   /**
    * The maximum size of each stream segments.
    *
+   * <p>Maximum size is {@link StreamCreator#MAX_SEGMENT_SIZE} (3 GB).
+   *
    * @param byteCapacity
    * @return this creator instance
    */

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 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").
@@ -46,6 +46,10 @@ public StreamCreator maxLengthBytes(ByteCapacity byteCapacity) {
 
   @Override
   public StreamCreator maxSegmentSizeBytes(ByteCapacity byteCapacity) {
+    if (byteCapacity != null && byteCapacity.compareTo(MAX_SEGMENT_SIZE) > 0) {
+      throw new IllegalArgumentException(
+          "The maximum segment size cannot be more than " + MAX_SEGMENT_SIZE);
+    }
     streamParametersBuilder.maxSegmentSizeBytes(byteCapacity);
     return this;
   }

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java

@@ -87,6 +87,8 @@
 import org.slf4j.LoggerFactory;
 import picocli.CommandLine;
 import picocli.CommandLine.Model.CommandSpec;
+import picocli.CommandLine.ParameterException;
+import picocli.CommandLine.Spec;
 
 @CommandLine.Command(
     name = "stream-perf-test",
@@ -105,6 +107,8 @@ public class StreamPerfTest implements Callable<Integer> {
       };
   private final String[] arguments;
 
+  @Spec CommandSpec spec; // injected by picocli
+
   @CommandLine.Mixin
   private final CommandLine.HelpCommand helpCommand = new CommandLine.HelpCommand();
 
@@ -220,12 +224,21 @@ public class StreamPerfTest implements Callable<Integer> {
       converter = Utils.ByteCapacityTypeConverter.class)
   private ByteCapacity maxLengthBytes;
 
+  private ByteCapacity maxSegmentSize;
+
   @CommandLine.Option(
       names = {"--stream-max-segment-size-bytes", "-smssb"},
       description = "max size of segments",
       defaultValue = "500mb",
       converter = Utils.ByteCapacityTypeConverter.class)
-  private ByteCapacity maxSegmentSize;
+  public void setMaxSegmentSize(ByteCapacity in) {
+    if (in != null && in.compareTo(StreamCreator.MAX_SEGMENT_SIZE) > 0) {
+      throw new ParameterException(
+          spec.commandLine(),
+          "The maximum segment size cannot be more than " + StreamCreator.MAX_SEGMENT_SIZE);
+    }
+    this.maxSegmentSize = in;
+  }
 
   @CommandLine.Option(
       names = {"--max-age", "-ma"},

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

@@ -0,0 +1,44 @@
+// Copyright (c) 2022 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").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.stream.impl;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.rabbitmq.stream.ByteCapacity;
+import org.junit.jupiter.api.Test;
+
+public class StreamCreatorTest {
+
+  @Test
+  void maxSegmentSizeBytesOK() {
+    new StreamStreamCreator(null)
+        .maxSegmentSizeBytes(ByteCapacity.MB(500))
+        .maxSegmentSizeBytes(ByteCapacity.MB(2999))
+        .maxSegmentSizeBytes(ByteCapacity.GB(3));
+  }
+
+  @Test
+  void maxSegmentSizeBytesKO() {
+    assertThatThrownBy(
+            () -> {
+              new StreamStreamCreator(null).maxSegmentSizeBytes(ByteCapacity.MB(3001));
+            })
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(
+            () -> {
+              new StreamStreamCreator(null).maxSegmentSizeBytes(ByteCapacity.GB(4));
+            })
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+}

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 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").
@@ -110,7 +110,7 @@ public static Duration waitAtMost(int timeoutInSeconds, CallableBooleanSupplier
     return waitAtMost(timeoutInSeconds, condition, null);
   }
 
-  static Duration waitAtMost(
+  public static Duration waitAtMost(
       int timeoutInSeconds, CallableBooleanSupplier condition, Supplier<String> message)
       throws Exception {
     if (condition.getAsBoolean()) {

src/test/java/com/rabbitmq/stream/perf/StreamPerfTestTest.java

@@ -101,7 +101,10 @@ void tearDownTest() {
   }
 
   private void waitRunEnds(int expectedExitCode) throws Exception {
-    waitAtMost(20, () -> exitCode.get() == expectedExitCode);
+    waitAtMost(
+        20,
+        () -> exitCode.get() == expectedExitCode,
+        () -> "Expected " + expectedExitCode + " exit code, got " + exitCode.get());
   }
 
   private void waitRunEnds() throws Exception {
@@ -266,7 +269,7 @@ void streamCreationIsIdempotentWhateverTheDifferencesInStreamProperties() throws
     Future<?> run =
         run(
             builder()
-                .maxLengthBytes(ByteCapacity.GB(42)) // different than already existing stream
+                .maxLengthBytes(ByteCapacity.GB(42)) // different from already existing stream
                 .streamMaxSegmentSizeBytes(ByteCapacity.MB(500))
                 .leaderLocator(LeaderLocator.LEAST_LEADERS));
     waitOneSecond();
@@ -275,6 +278,14 @@ void streamCreationIsIdempotentWhateverTheDifferencesInStreamProperties() throws
     assertThat(consoleOutput()).contains("Warning: stream '" + s + "'");
   }
 
+  @Test
+  void exceedingMaxSegmentSizeLimitShouldGenerateError() throws Exception {
+    run(builder().streamMaxSegmentSizeBytes(ByteCapacity.TB(1)));
+    waitRunEnds(2);
+    assertThat(consoleErrorOutput()).contains("The maximum segment size cannot be more than 3GB");
+    checkErrIsEmpty = false;
+  }
+
   @Test
   void byteRatesShouldBeIncludedWhenOptionIsEnabled() throws Exception {
     Future<?> run = run(builder().byteRates());