Connect to any stream member for consumers when using a load balancer

.github/workflows/test-pr.yml

@@ -33,10 +33,16 @@ jobs:
             -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem
       - name: Test (dynamic-batch publishing)
         run: |
-          ./mvnw verify -Drabbitmqctl.bin=DOCKER:rabbitmq \
+          ./mvnw test -Drabbitmqctl.bin=DOCKER:rabbitmq \
             -Drabbitmq.stream.producer.dynamic.batch=true \
             -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
             -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
             -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem
       - name: Stop broker
         run: docker stop rabbitmq && docker rm rabbitmq
+      - name: Start cluster
+        run: ci/start-cluster.sh
+      - name: Test against cluster
+        run: ./mvnw test -Dtest="*ClusterTest" -Drabbitmqctl.bin=DOCKER:rabbitmq0
+      - name: Stop cluster
+        run: docker compose --file ci/cluster/docker-compose.yml down

ci/cluster/configuration-0/enabled_plugins

@@ -0,0 +1 @@
+[rabbitmq_stream_management].

ci/cluster/configuration-0/rabbitmq.conf

@@ -0,0 +1,8 @@
+cluster_formation.peer_discovery_backend  = rabbit_peer_discovery_classic_config
+cluster_formation.classic_config.nodes.1 = rabbit@node0
+cluster_formation.classic_config.nodes.2 = rabbit@node1
+cluster_formation.classic_config.nodes.3 = rabbit@node2
+loopback_users = none
+
+stream.advertised_host = localhost
+stream.advertised_port = 5552

ci/cluster/configuration-1/enabled_plugins

@@ -0,0 +1 @@
+[rabbitmq_stream_management].

ci/cluster/configuration-1/rabbitmq.conf

@@ -0,0 +1,8 @@
+cluster_formation.peer_discovery_backend  = rabbit_peer_discovery_classic_config
+cluster_formation.classic_config.nodes.1 = rabbit@node0
+cluster_formation.classic_config.nodes.2 = rabbit@node1
+cluster_formation.classic_config.nodes.3 = rabbit@node2
+loopback_users = none
+
+stream.advertised_host = localhost
+stream.advertised_port = 5553

ci/cluster/configuration-2/enabled_plugins

@@ -0,0 +1 @@
+[rabbitmq_stream_management].

ci/cluster/configuration-2/rabbitmq.conf

@@ -0,0 +1,8 @@
+cluster_formation.peer_discovery_backend  = rabbit_peer_discovery_classic_config
+cluster_formation.classic_config.nodes.1 = rabbit@node0
+cluster_formation.classic_config.nodes.2 = rabbit@node1
+cluster_formation.classic_config.nodes.3 = rabbit@node2
+loopback_users = none
+
+stream.advertised_host = localhost
+stream.advertised_port = 5554

ci/cluster/docker-compose.yml

@@ -0,0 +1,64 @@
+services:
+  node0:
+    environment:
+      - RABBITMQ_ERLANG_COOKIE='secret_cookie'
+    networks:
+      - rabbitmq-cluster
+    hostname: node0
+    container_name: rabbitmq0
+    image: ${RABBITMQ_IMAGE:-rabbitmq:4.0}
+    pull_policy: always
+    ports:
+      - "5672:5672"
+      - "5552:5552"
+      - "15672:15672"
+    tty: true
+    volumes:
+      - ./configuration-0/:/etc/rabbitmq/
+  node1:
+    environment:
+      - RABBITMQ_ERLANG_COOKIE='secret_cookie'
+    networks:
+      - rabbitmq-cluster
+    hostname: node1
+    container_name: rabbitmq1
+    image: ${RABBITMQ_IMAGE:-rabbitmq:4.0}
+    pull_policy: always
+    ports:
+      - "5673:5672"
+      - "5553:5552"
+      - "15673:15672"
+    tty: true
+    volumes:
+      - ./configuration-1/:/etc/rabbitmq/
+  node2:
+    environment:
+      - RABBITMQ_ERLANG_COOKIE='secret_cookie'
+    networks:
+      - rabbitmq-cluster
+    hostname: node2
+    container_name: rabbitmq2
+    image: ${RABBITMQ_IMAGE:-rabbitmq:4.0}
+    pull_policy: always
+    ports:
+      - "5674:5672"
+      - "5554:5552"
+      - "15674:15672"
+    tty: true
+    volumes:
+      - ./configuration-2/:/etc/rabbitmq/
+  load-balander:
+    networks:
+      - rabbitmq-cluster
+    hostname: load-balancer
+    container_name: haproxy
+    image: haproxy:3.0
+    pull_policy: always
+    ports:
+      - "5555:5555"
+      - "8100:8100"
+    tty: true
+    volumes:
+      - ./load-balancer/:/usr/local/etc/haproxy:ro
+networks:
+  rabbitmq-cluster:

ci/cluster/load-balancer/haproxy.cfg

@@ -0,0 +1,31 @@
+global
+        log 127.0.0.1   local0 info
+        maxconn 512
+
+defaults
+        log     global
+        mode    tcp
+        option  tcplog
+        option  dontlognull
+        retries 3
+        option redispatch
+        maxconn 512
+        timeout connect 5s
+        timeout client 120s
+        timeout server 120s
+
+listen stream
+        bind :5555
+        mode tcp
+        balance roundrobin
+        server rabbitmq-0 node0:5552
+        server rabbitmq-1 node1:5552
+        server rabbitmq-2 node2:5552
+
+listen  stats
+        bind :8100
+        mode http
+        option httplog
+        stats enable
+        stats uri       /stats
+        stats refresh 5s

ci/start-cluster.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+
+export RABBITMQ_IMAGE=${RABBITMQ_IMAGE:-rabbitmq:4.0}
+
+wait_for_message() {
+  while ! docker logs "$1" | grep -q "$2";
+  do
+      sleep 2
+      echo "Waiting 2 seconds for $1 to start..."
+  done
+}
+
+docker compose --file ci/cluster/docker-compose.yml down
+docker compose --file ci/cluster/docker-compose.yml up --detach
+
+wait_for_message rabbitmq0 "completed with"
+
+docker exec rabbitmq0 rabbitmqctl await_online_nodes 3
+
+docker exec rabbitmq0 rabbitmqctl enable_feature_flag --opt-in khepri_db
+docker exec rabbitmq1 rabbitmqctl enable_feature_flag --opt-in khepri_db
+docker exec rabbitmq2 rabbitmqctl enable_feature_flag --opt-in khepri_db
+
+docker exec rabbitmq0 rabbitmqctl cluster_status
+
+docker compose --file ci/cluster/docker-compose.yml ps