<p>I deployed a Kubernetes cluster using the RabbitMQ Operator and activated the <code>rabbitmq_stream</code> plugin. This is my yaml:</p>
<pre class="lang-yaml prettyprint-override"><code>apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
  name: rabbitmq-deployment
  namespace: rabbitmq-namespace
spec:
  replicas: 2
  image: rabbitmq:3.11.13
  persistence:
    storage: 20Gi
  service:
    type: LoadBalancer
  rabbitmq:
    additionalPlugins:
      - rabbitmq_stream
      - rabbitmq_stream_management
</code></pre>
<p>Also i use the RabbitMQ Java Stream Client and i'm connecting to the cluster like this:</p>
<pre class="lang-java prettyprint-override"><code>EnvironmentBuilder environmentBuilder = Environment.builder();
environmentBuilder.host(System.getenv(&quot;RABBITMQ_HOST&quot;));
environmentBuilder.port(Integer.parseInt(System.getenv(&quot;RABBITMQ_STREAM_PORT&quot;)));
environmentBuilder.username(System.getenv(&quot;RABBITMQ_USERNAME&quot;));
environmentBuilder.password(System.getenv(&quot;RABBITMQ_PASSWORD&quot;));
mainConnection = environmentBuilder.build();
</code></pre>
<p>Now when i use this client so create the stream, it's working flawlessy and no error is reported:</p>
<pre class="lang-java prettyprint-override"><code>mainConnection.streamCreator().stream(&quot;mystream&quot;).maxAge(Duration.of(1, ChronoUnit.DAYS)).create()
</code></pre>
<p>Now when i try to produce messages like this:</p>
<pre class="lang-java prettyprint-override"><code>Producer producer = RabbitMQStreamConnection.mainConnection.producerBuilder().stream(&quot;mystream&quot;).build();
byte[] messagePayload = &quot;hello&quot;.getBytes(StandardCharsets.UTF_8);
producer.send(
    producer.messageBuilder().addData(messagePayload).build(),
    confirmationStatus -&gt; {
        if (confirmationStatus.isConfirmed()) {
            // the message made it to the broker
        } else {
            // the message did not make it to the broker
        }
});
</code></pre>
<p>It throws this exception:</p>
<pre><code>com.rabbitmq.stream.StreamException
Error while creating stream connection to rabbitmq-deployment-server-0.rabbitmq-deployment-nodes.rabbitmq-namespace:5552
</code></pre>
<p>Of course, because there are two nodes (<code>replicas = 2</code>) and it seems like traffic gets redirected directly.</p>
<p>What i want is that i can produce &amp; consume messages from the stream.</p>
<p>Right now, i have no clue what i could do next to solve this problem.</p>

Stefan
<p>You should use the load balancer configuration.</p>
<p>See: <a href="https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/#when-a-load-balancer-is-in-use" rel="nofollow noreferrer">https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/#when-a-load-balancer-is-in-use</a></p>
<blockquote>
<p>A load balancer can misguide the client when it tries to connect to nodes that host stream leaders and replicas. The &quot;Connecting to Streams&quot; blog post covers why client applications must connect to the appropriate nodes in a cluster and how a load balancer can make things complicated for them.</p>
</blockquote>
<blockquote>
<p>The EnvironmentBuilder#addressResolver(AddressResolver) method allows intercepting the node resolution after metadata hints and before connection. Applications can use this hook to ignore metadata hints and always use the load balancer, as illustrated in the following snippet:</p>
</blockquote>
<p>Using a custom address resolver to always use a load balancer</p>
<pre class="lang-java prettyprint-override"><code>Address entryPoint = new Address(&quot;my-load-balancer&quot;, 5552);  
Environment environment = Environment.builder()
    .host(entryPoint.host())  
    .port(entryPoint.port())  
    .addressResolver(address -&gt; entryPoint)  
    .build();
</code></pre>
