Force localhost for local development

src/docs/asciidoc/api.adoc

@@ -77,6 +77,11 @@ Creating publishers and consumers can cause problems as the client uses hints fr
 These connection hints can be accurate or less appropriate depending on the infrastructure.
 If you hit some connection problems at some point – like hostnames impossible to resolve for client applications - this https://blog.rabbitmq.com/posts/2021/07/connecting-to-streams/[blog post] should help you understand what is going on and fix the issues.
 
+To make the local development experience simple, the client library can choose to always use `localhost` for producers and consumers.
+This happens if the following conditions are met: the initial host to connect to is `localhost`, the user is `guest`, and no custom address resolver has been provided.
+Provide a pass-through `AddressResolver` to `EnvironmentBuilder#addressResolver(AddressResolver)` to avoid this behavior.
+It is unlikely this behavior applies for any real-world deployment, where `localhost` and/or the default `guest` user should not be used.
+
 ===== Enabling TLS
 
 TLS can be enabled by using the `rabbitmq-stream+tls` scheme in the URI.

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

@@ -63,8 +63,26 @@ public interface EnvironmentBuilder {
    * <p>Applications can use this abstraction to make sure connection attempts ignore metadata hints
    * and always go to a single point like a load balancer.
    *
+   * <p>The default implementation does not perform any logic, it just returns the passed-in
+   * address.
+   *
+   * <p><i>The default implementation is overridden automatically if the following conditions are
+   * met: the host to connect to is <code>localhost</code>, the user is <code>guest</code>, and no
+   * address resolver has been provided. The client will then always tries to connect to <code>
+   * localhost</code> to facilitate the so-called "beginner experience", that is a developer working
+   * against a local RabbitMQ instance. Just provide a pass-through address resolver to avoid this
+   * behavior, e.g.:</i>
+   *
+   * <pre>
+   * Environment.builder()
+   *   .addressResolver(address -> address)
+   *   .build();
+   * </pre>
+   *
    * @param addressResolver
    * @return this builder instance
+   * @see <a href="https://blog.rabbitmq.com/posts/2021/07/connecting-to-streams/">"Connecting to
+   *     Streams" blog post</a>
    */
   EnvironmentBuilder addressResolver(AddressResolver addressResolver);
 
@@ -172,7 +190,7 @@ public interface EnvironmentBuilder {
    *
    * @param requestedHeartbeat
    * @return this builder instance
-   * @see <a href="https://rabbitmq.com/stream.html#protocol">See stream plugin documentation</a>
+   * @see <a href="https://rabbitmq.com/stream.html#protocol">Stream plugin documentation</a>
    */
   EnvironmentBuilder requestedHeartbeat(Duration requestedHeartbeat);
 
@@ -183,7 +201,7 @@ public interface EnvironmentBuilder {
    *
    * @param requestedMaxFrameSize
    * @return this builder instance
-   * @see <a href="https://rabbitmq.com/stream.html#protocol">See stream plugin documentation</a>
+   * @see <a href="https://rabbitmq.com/stream.html#protocol">Stream plugin documentation</a>
    */
   EnvironmentBuilder requestedMaxFrameSize(int requestedMaxFrameSize);
 

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

@@ -14,11 +14,7 @@
 package com.rabbitmq.stream.impl;
 
 import static com.rabbitmq.stream.Constants.*;
-import static com.rabbitmq.stream.impl.Utils.encodeRequestCode;
-import static com.rabbitmq.stream.impl.Utils.encodeResponseCode;
-import static com.rabbitmq.stream.impl.Utils.extractResponseCode;
-import static com.rabbitmq.stream.impl.Utils.formatConstant;
-import static com.rabbitmq.stream.impl.Utils.noOpConsumer;
+import static com.rabbitmq.stream.impl.Utils.*;
 import static java.lang.String.format;
 import static java.lang.String.join;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -495,10 +491,21 @@ private void authenticate() {
       } else if (saslAuthenticateResponse.isChallenge()) {
         challenge = saslAuthenticateResponse.challenge;
       } else if (saslAuthenticateResponse.isAuthenticationFailure()) {
-        throw new AuthenticationFailureException(
+        String message =
             "Unexpected response code during authentication: "
-                + formatConstant(saslAuthenticateResponse.getResponseCode()),
-            saslAuthenticateResponse.getResponseCode());
+                + formatConstant(saslAuthenticateResponse.getResponseCode());
+        if (saslAuthenticateResponse.getResponseCode()
+            == RESPONSE_CODE_AUTHENTICATION_FAILURE_LOOPBACK) {
+          message +=
+              ". The user is not authorized to connect from a remote host. "
+                  + "If the broker is running locally, make sure the '"
+                  + this.host
+                  + "' hostname is resolved to "
+                  + "the loopback interface (localhost, 127.0.0.1, ::1). "
+                  + "See https://www.rabbitmq.com/access-control.html#loopback-users.";
+        }
+        throw new AuthenticationFailureException(
+            message, saslAuthenticateResponse.getResponseCode());
       } else {
         throw new StreamException(
             "Unexpected response code during authentication: "
@@ -2234,7 +2241,7 @@ public static class ClientParameters {
     private ShutdownListener shutdownListener = shutdownContext -> {};
     private SaslConfiguration saslConfiguration = DefaultSaslConfiguration.PLAIN;
     private CredentialsProvider credentialsProvider =
-        new DefaultUsernamePasswordCredentialsProvider("guest", "guest");
+        new DefaultUsernamePasswordCredentialsProvider(DEFAULT_USERNAME, "guest");
     private ChunkChecksum chunkChecksum = JdkChunkChecksum.CRC32_SINGLETON;
     private MetricsCollector metricsCollector = NoOpMetricsCollector.SINGLETON;
     private SslContext sslContext;
@@ -2438,6 +2445,10 @@ Codec codec() {
       return this.codec;
     }
 
+    CredentialsProvider credentialsProvider() {
+      return this.credentialsProvider;
+    }
+
     public ClientParameters channelCustomizer(Consumer<Channel> channelCustomizer) {
       this.channelCustomizer = channelCustomizer;
       return this;

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

@@ -14,10 +14,7 @@
 package com.rabbitmq.stream.impl;
 
 import static com.rabbitmq.stream.impl.AsyncRetry.asyncRetry;
-import static com.rabbitmq.stream.impl.Utils.convertCodeToException;
-import static com.rabbitmq.stream.impl.Utils.exceptionMessage;
-import static com.rabbitmq.stream.impl.Utils.formatConstant;
-import static com.rabbitmq.stream.impl.Utils.namedRunnable;
+import static com.rabbitmq.stream.impl.Utils.*;
 import static java.lang.String.format;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
@@ -31,6 +28,8 @@
 import com.rabbitmq.stream.impl.StreamConsumerBuilder.TrackingConfiguration;
 import com.rabbitmq.stream.impl.StreamEnvironmentBuilder.DefaultTlsConfiguration;
 import com.rabbitmq.stream.impl.Utils.ClientConnectionType;
+import com.rabbitmq.stream.sasl.CredentialsProvider;
+import com.rabbitmq.stream.sasl.UsernamePasswordCredentialsProvider;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -109,7 +108,6 @@ class StreamEnvironment implements Environment {
     clientParametersPrototype = clientParametersPrototype.byteBufAllocator(byteBufAllocator);
     clientParametersPrototype = maybeSetUpClientParametersFromUris(uris, clientParametersPrototype);
 
-    this.addressResolver = addressResolver;
     this.observationCollector = observationCollector;
 
     boolean tls;
@@ -148,6 +146,26 @@ class StreamEnvironment implements Environment {
               .collect(Collectors.toList());
     }
 
+    AddressResolver addressResolverToUse = addressResolver;
+    if (this.addresses.size() == 1
+        && "localhost".equals(this.addresses.get(0).host())
+        && addressResolver == DEFAULT_ADDRESS_RESOLVER) {
+      CredentialsProvider credentialsProvider = clientParametersPrototype.credentialsProvider();
+      if (credentialsProvider instanceof UsernamePasswordCredentialsProvider) {
+        String username = ((UsernamePasswordCredentialsProvider) credentialsProvider).getUsername();
+        if (DEFAULT_USERNAME.equals(username)) {
+          Address address = new Address("localhost", clientParametersPrototype.port());
+          addressResolverToUse = ignored -> address;
+          LOGGER.info(
+              "Connecting to localhost with {} user, assuming development environment",
+              DEFAULT_USERNAME);
+          LOGGER.info("Using address resolver to always connect to localhost");
+        }
+      }
+    }
+
+    this.addressResolver = addressResolverToUse;
+
     this.addresses.forEach(address -> this.locators.add(new Locator(address)));
     this.executorServiceFactory =
         new DefaultExecutorServiceFactory(

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

@@ -13,6 +13,7 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
+import static com.rabbitmq.stream.impl.Utils.DEFAULT_ADDRESS_RESOLVER;
 import static com.rabbitmq.stream.impl.Utils.noOpConsumer;
 
 import com.rabbitmq.stream.*;
@@ -55,7 +56,7 @@ public class StreamEnvironmentBuilder implements EnvironmentBuilder {
       BackOffDelayPolicy.fixed(Duration.ofSeconds(5));
   private BackOffDelayPolicy topologyBackOffDelayPolicy =
       BackOffDelayPolicy.fixedWithInitialDelay(Duration.ofSeconds(5), Duration.ofSeconds(1));
-  private AddressResolver addressResolver = address -> address;
+  private AddressResolver addressResolver = DEFAULT_ADDRESS_RESOLVER;
   private int maxProducersByConnection = ProducersCoordinator.MAX_PRODUCERS_PER_CLIENT;
   private int maxTrackingConsumersByConnection =
       ProducersCoordinator.MAX_TRACKING_CONSUMERS_PER_CLIENT;

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

@@ -15,14 +15,7 @@
 
 import static java.lang.String.format;
 
-import com.rabbitmq.stream.Address;
-import com.rabbitmq.stream.BackOffDelayPolicy;
-import com.rabbitmq.stream.Constants;
-import com.rabbitmq.stream.ConsumerUpdateListener;
-import com.rabbitmq.stream.OffsetSpecification;
-import com.rabbitmq.stream.StreamDoesNotExistException;
-import com.rabbitmq.stream.StreamException;
-import com.rabbitmq.stream.StreamNotAvailableException;
+import com.rabbitmq.stream.*;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
 import java.net.UnknownHostException;
 import java.security.cert.X509Certificate;
@@ -87,6 +80,9 @@ final class Utils {
     CONSTANT_LABELS = Collections.unmodifiableMap(labels);
   }
 
+  static final AddressResolver DEFAULT_ADDRESS_RESOLVER = address -> address;
+  static final String DEFAULT_USERNAME = "guest";
+
   private Utils() {}
 
   @SuppressWarnings("unchecked")

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

@@ -693,7 +693,7 @@ public Integer call() throws Exception {
           closeStep("Closing environment executor", () -> envExecutor.shutdownNow()));
 
       boolean tls = isTls(this.uris);
-      AddressResolver addrResolver;
+      AddressResolver addrResolver = null;
       if (loadBalancer) {
         int defaultPort = tls ? Client.DEFAULT_TLS_PORT : Client.DEFAULT_PORT;
         List<Address> addresses =
@@ -717,9 +717,7 @@ public Integer call() throws Exception {
         addrResolver =
             address -> addresses.get(connectionAttemptCount.getAndIncrement() % addresses.size());
       } else {
-        if (this.addressResolver == null) {
-          addrResolver = address -> address;
-        } else {
+        if (this.addressResolver != null) {
           addrResolver = this.addressResolver; // should happen only in tests
         }
       }
@@ -737,7 +735,6 @@ public Integer call() throws Exception {
           Environment.builder()
               .id("stream-perf-test")
               .uris(this.uris)
-              .addressResolver(addrResolver)
               .scheduledExecutorService(envExecutor)
               .metricsCollector(metricsCollector)
               .netty()
@@ -752,6 +749,10 @@ public Integer call() throws Exception {
               .rpcTimeout(Duration.ofSeconds(this.rpcTimeout))
               .requestedMaxFrameSize((int) this.requestedMaxFrameSize.toBytes());
 
+      if (addrResolver != null) {
+        environmentBuilder = environmentBuilder.addressResolver(addrResolver);
+      }
+
       java.util.function.Consumer<io.netty.channel.Channel> channelCustomizer = channel -> {};
 
       if (tls) {

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

@@ -17,7 +17,6 @@
 import static com.rabbitmq.stream.Host.memoryAlarm;
 import static com.rabbitmq.stream.impl.TestUtils.ExceptionConditions.responseCode;
 import static com.rabbitmq.stream.impl.TestUtils.latchAssert;
-import static com.rabbitmq.stream.impl.TestUtils.localhost;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static java.util.stream.IntStream.range;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -67,7 +66,6 @@ static void afterAll() throws Exception {
   @BeforeEach
   void init() {
     environmentBuilder = Environment.builder();
-    environmentBuilder.addressResolver(add -> localhost());
     env = environmentBuilder.netty().eventLoopGroup(eventLoopGroup).environmentBuilder().build();
   }
 

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

@@ -61,7 +61,6 @@ public class FilteringTest {
   void init() throws Exception {
     EnvironmentBuilder environmentBuilder =
         Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
-    environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
   }
 

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

@@ -67,7 +67,6 @@ static void afterAll() throws Exception {
   @BeforeEach
   void init() {
     environmentBuilder = Environment.builder();
-    environmentBuilder.addressResolver(add -> localhost());
     env = environmentBuilder.netty().eventLoopGroup(eventLoopGroup).environmentBuilder().build();
   }
 

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

@@ -13,7 +13,6 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
-import static com.rabbitmq.stream.impl.TestUtils.localhost;
 import static com.rabbitmq.stream.impl.TestUtils.publishAndWaitForConfirms;
 import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -53,7 +52,6 @@ void init() {
             .eventLoopGroup(eventLoopGroup)
             .environmentBuilder()
             .maxConsumersByConnection(1);
-    environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
   }
 

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

@@ -15,7 +15,6 @@
 
 import static com.rabbitmq.stream.impl.TestUtils.declareSuperStreamTopology;
 import static com.rabbitmq.stream.impl.TestUtils.deleteSuperStreamTopology;
-import static com.rabbitmq.stream.impl.TestUtils.localhost;
 import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static java.util.stream.Collectors.toList;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -66,7 +65,6 @@ public class SacSuperStreamConsumerTest {
   void init(TestInfo info) throws Exception {
     EnvironmentBuilder environmentBuilder =
         Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
-    environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
     superStream = TestUtils.streamName(info);
     connection = new ConnectionFactory().newConnection();

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

@@ -107,7 +107,6 @@ static long offset(String line) {
   @BeforeEach
   void init() throws Exception {
     environmentBuilder = Environment.builder();
-    environmentBuilder.addressResolver(add -> localhost());
     env = environmentBuilder.netty().eventLoopGroup(eventLoopGroup).environmentBuilder().build();
     socket = new Socket("localhost", 61613);
     out = socket.getOutputStream();

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

@@ -96,7 +96,6 @@ void init() {
       recoveryInitialDelay = RECOVERY_DELAY;
     }
     EnvironmentBuilder environmentBuilder = environmentBuilder();
-    environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
   }
 

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

@@ -70,7 +70,6 @@ public class StreamProducerTest {
   @BeforeEach
   void init() {
     EnvironmentBuilder environmentBuilder = environmentBuilder();
-    environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
   }
 

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

@@ -18,7 +18,6 @@
 import static com.rabbitmq.stream.impl.TestUtils.declareSuperStreamTopology;
 import static com.rabbitmq.stream.impl.TestUtils.deleteSuperStreamTopology;
 import static com.rabbitmq.stream.impl.TestUtils.latchAssert;
-import static com.rabbitmq.stream.impl.TestUtils.localhost;
 import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static com.rabbitmq.stream.impl.TestUtils.wrap;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -63,7 +62,6 @@ public class SuperStreamConsumerTest {
   void init(TestInfo info) throws Exception {
     EnvironmentBuilder environmentBuilder =
         Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
-    environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
     superStream = TestUtils.streamName(info);
     connection = new ConnectionFactory().newConnection();

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

@@ -16,7 +16,6 @@
 import static com.rabbitmq.stream.impl.TestUtils.declareSuperStreamTopology;
 import static com.rabbitmq.stream.impl.TestUtils.deleteSuperStreamTopology;
 import static com.rabbitmq.stream.impl.TestUtils.latchAssert;
-import static com.rabbitmq.stream.impl.TestUtils.localhost;
 import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -61,7 +60,6 @@ public class SuperStreamProducerTest {
   void init(TestInfo info) throws Exception {
     EnvironmentBuilder environmentBuilder =
         Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
-    environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
     connection = new ConnectionFactory().newConnection();
     superStream = TestUtils.streamName(info);

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

@@ -16,7 +16,6 @@
 import static com.rabbitmq.stream.impl.TestUtils.declareSuperStreamTopology;
 import static com.rabbitmq.stream.impl.TestUtils.deleteSuperStreamTopology;
 import static com.rabbitmq.stream.impl.TestUtils.latchAssert;
-import static com.rabbitmq.stream.impl.TestUtils.localhost;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.rabbitmq.client.Connection;
@@ -63,7 +62,6 @@ public class SuperStreamTest {
   void init(TestInfo info) throws Exception {
     EnvironmentBuilder environmentBuilder =
         Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
-    environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
     connection = new ConnectionFactory().newConnection();
     superStream = TestUtils.streamName(info);