Use number of available processors for stream count in test

Instead of 10. This decreases load for low-resource environments,
like CI (usually ~ 4 available processors). The VM has to run the test
suite itself, but also the 3-node cluster.

Author: acogoluegnes

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

@@ -75,7 +75,7 @@ final class ConsumersCoordinator implements AutoCloseable {
   private final List<SubscriptionTracker> trackers = new CopyOnWriteArrayList<>();
   private final ExecutorServiceFactory executorServiceFactory =
       new DefaultExecutorServiceFactory(
-          Runtime.getRuntime().availableProcessors(), 10, "rabbitmq-stream-consumer-connection-");
+          AVAILABLE_PROCESSORS, 10, "rabbitmq-stream-consumer-connection-");
   private final boolean forceReplica;
   private final Lock coordinatorLock = new ReentrantLock();
 

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

@@ -68,7 +68,7 @@ final class ProducersCoordinator implements AutoCloseable {
   private final List<ProducerTracker> producerTrackers = new CopyOnWriteArrayList<>();
   private final ExecutorServiceFactory executorServiceFactory =
       new DefaultExecutorServiceFactory(
-          Runtime.getRuntime().availableProcessors(), 10, "rabbitmq-stream-producer-connection-");
+          AVAILABLE_PROCESSORS, 10, "rabbitmq-stream-producer-connection-");
   private final Lock coordinatorLock = new ReentrantLock();
   private final boolean forceLeader;
 
@@ -750,7 +750,10 @@ private void assignProducersToNewManagers(
                 List<BrokerWrapper> candidates = brokerAndCandidates.v2();
                 String key = keyForNode(broker);
                 LOGGER.debug(
-                    "Assigning {} producer(s) and consumer tracker(s) to {} (stream '{}')", trackers.size(), key, stream);
+                    "Assigning {} producer(s) and consumer tracker(s) to {} (stream '{}')",
+                    trackers.size(),
+                    key,
+                    stream);
                 trackers.forEach(tracker -> maybeRecoverAgent(broker, candidates, tracker));
               })
           .exceptionally(

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

@@ -224,7 +224,7 @@ class StreamEnvironment implements Environment {
     }
     ScheduledExecutorService executorService;
     if (scheduledExecutorService == null) {
-      int threads = Runtime.getRuntime().availableProcessors();
+      int threads = AVAILABLE_PROCESSORS;
       LOGGER.debug("Creating scheduled executor service with {} thread(s)", threads);
       ThreadFactory threadFactory = threadFactory("rabbitmq-stream-environment-scheduler-");
       executorService = Executors.newScheduledThreadPool(threads, threadFactory);

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

@@ -41,6 +41,8 @@
 
 final class Utils {
 
+  static final int AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();
+
   @SuppressWarnings("rawtypes")
   private static final Consumer NO_OP_CONSUMER = o -> {};
 

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

@@ -28,6 +28,7 @@
 import com.google.common.collect.Streams;
 import com.google.common.util.concurrent.RateLimiter;
 import com.rabbitmq.stream.*;
+import com.rabbitmq.stream.impl.TestUtils.DisabledIfNotCluster;
 import com.rabbitmq.stream.impl.TestUtils.Sync;
 import com.rabbitmq.stream.impl.Tuples.Pair;
 import io.netty.channel.ChannelOption;
@@ -52,7 +53,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-@TestUtils.DisabledIfNotCluster
+@DisabledIfNotCluster
 @StreamTestInfrastructure
 public class RecoveryClusterTest {
 
@@ -87,7 +88,7 @@ static void initAll() {
 
   @BeforeEach
   void init(TestInfo info) {
-    int availableProcessors = Runtime.getRuntime().availableProcessors();
+    int availableProcessors = Utils.AVAILABLE_PROCESSORS;
     LOGGER.info("Available processors: {}", availableProcessors);
     ThreadFactory threadFactory = threadFactory("rabbitmq-stream-environment-scheduler-");
     scheduledExecutorService = Executors.newScheduledThreadPool(availableProcessors, threadFactory);
@@ -134,7 +135,7 @@ void clusterRestart(boolean useLoadBalancer, boolean forceLeader) throws Interru
         "Cluster restart test, use load balancer {}, force leader {}",
         useLoadBalancer,
         forceLeader);
-    int streamCount = 10;
+    int streamCount = Utils.AVAILABLE_PROCESSORS;
     int producerCount = streamCount * 2;
     int consumerCount = streamCount * 2;
 

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

@@ -17,7 +17,6 @@
 import static com.rabbitmq.stream.ConsumerFlowStrategy.creditWhenHalfMessagesProcessed;
 import static com.rabbitmq.stream.impl.TestUtils.*;
 import static com.rabbitmq.stream.impl.TestUtils.CountDownLatchConditions.completed;
-import static java.lang.Runtime.getRuntime;
 import static java.lang.String.format;
 import static java.util.Collections.synchronizedList;
 import static org.assertj.core.api.Assertions.*;
@@ -243,8 +242,7 @@ void consumeWithAsyncConsumerFlowControl() throws Exception {
   void asynchronousProcessingWithFlowControl() {
     int messageCount = 100_000;
     publishAndWaitForConfirms(cf, messageCount, stream);
-    ExecutorService executorService =
-        Executors.newFixedThreadPool(getRuntime().availableProcessors());
+    ExecutorService executorService = Executors.newFixedThreadPool(Utils.AVAILABLE_PROCESSORS);
     try {
       CountDownLatch latch = new CountDownLatch(messageCount);
       environment.consumerBuilder().stream(stream)