[revert] Revert "[improve][broker] Support showing client ip address in client stats while using reverse proxy (#23974)" (#24177)

Author: lhotari

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

@@ -48,7 +48,6 @@
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.PositionFactory;
 import org.apache.bookkeeper.mledger.impl.AckSetStateUtil;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.mutable.MutableInt;
 import org.apache.commons.lang3.tuple.MutablePair;
 import org.apache.commons.lang3.tuple.Pair;
@@ -65,7 +64,6 @@
 import org.apache.pulsar.common.api.proto.KeyLongValue;
 import org.apache.pulsar.common.api.proto.KeySharedMeta;
 import org.apache.pulsar.common.api.proto.MessageIdData;
-import org.apache.pulsar.common.naming.Metadata;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.ClusterPolicies.ClusterUrl;
 import org.apache.pulsar.common.policies.data.TopicOperation;
@@ -228,8 +226,7 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo
         this.metadata = metadata != null ? metadata : Collections.emptyMap();
 
         stats = new ConsumerStatsImpl();
-        String address = metadata != null ? metadata.get(Metadata.CLIENT_IP) : null;
-        stats.setAddress(StringUtils.isNotBlank(address) ? address : cnx.clientSourceAddressAndPort());
+        stats.setAddress(cnx.clientSourceAddressAndPort());
         stats.consumerName = consumerName;
         stats.appId = appId;
         stats.setConnectedSince(DateFormatter.format(connectedSince));

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java

@@ -42,7 +42,6 @@
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import org.apache.bookkeeper.mledger.Position;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.intercept.BrokerInterceptor;
 import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
@@ -56,7 +55,6 @@
 import org.apache.pulsar.common.api.proto.MessageMetadata;
 import org.apache.pulsar.common.api.proto.ProducerAccessMode;
 import org.apache.pulsar.common.api.proto.ServerError;
-import org.apache.pulsar.common.naming.Metadata;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.ClusterPolicies.ClusterUrl;
 import org.apache.pulsar.common.policies.data.TopicOperation;
@@ -133,8 +131,7 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN
         this.metadata = metadata != null ? metadata : Collections.emptyMap();
 
         this.stats = isNonPersistentTopic ? new NonPersistentPublisherStatsImpl() : new PublisherStatsImpl();
-        String address = metadata != null ? metadata.get(Metadata.CLIENT_IP) : null;
-        stats.setAddress(StringUtils.isNotBlank(address) ? address : cnx.clientSourceAddressAndPort());
+        stats.setAddress(cnx.clientSourceAddressAndPort());
         stats.setConnectedSince(DateFormatter.now());
         stats.setClientVersion(cnx.getClientVersion());
         stats.setProducerName(producerName);

pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsSniTest.java

@@ -18,19 +18,16 @@
  */
 package org.apache.pulsar.client.api;
 
-import static org.testng.Assert.assertNotNull;
 import java.net.InetAddress;
 import java.net.URI;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+
 import org.apache.pulsar.client.impl.auth.AuthenticationTls;
 import org.testng.annotations.Test;
+
 import lombok.Cleanup;
-import org.apache.pulsar.broker.service.persistent.PersistentTopic;
-import org.apache.pulsar.broker.service.Consumer;
-import org.apache.pulsar.broker.service.Producer;
-import org.apache.pulsar.common.naming.Metadata;
 
 @Test(groups = "broker-api")
 public class TlsSniTest extends TlsProducerConsumerBase {
@@ -54,7 +51,6 @@ public void testIpAddressInBrokerServiceUrl() throws Exception {
 
         ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(brokerServiceIpAddressUrl)
                 .tlsTrustCertsFilePath(CA_CERT_FILE_PATH).allowTlsInsecureConnection(false)
-                .proxyServiceUrl(brokerServiceIpAddressUrl, ProxyProtocol.SNI)
                 .enableTlsHostnameVerification(false)
                 .operationTimeout(1000, TimeUnit.MILLISECONDS);
         Map<String, String> authParams = new HashMap<>();
@@ -66,12 +62,6 @@ public void testIpAddressInBrokerServiceUrl() throws Exception {
         PulsarClient pulsarClient = clientBuilder.build();
         // should be able to create producer successfully
         pulsarClient.newProducer().topic(topicName).create();
-        pulsarClient.newConsumer().topic(topicName).subscriptionName("test").subscribe();
-        PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get();
-        Producer producer = topic.getProducers().values().iterator().next();
-        assertNotNull(producer.getMetadata().get(Metadata.CLIENT_IP));
-        Consumer consumer = topic.getSubscription("test").getDispatcher().getConsumers().iterator().next();
-        assertNotNull(consumer.getMetadata().get(Metadata.CLIENT_IP));
     }
 }
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

@@ -119,7 +119,6 @@
 @SuppressWarnings("unchecked")
 public class ClientCnx extends PulsarHandler {
 
-    private static final Logger log = LoggerFactory.getLogger(ClientCnx.class);
     protected final Authentication authentication;
     protected State state;
 
@@ -1439,13 +1438,7 @@ private void checkRequestTimeout() {
         }
     }
 
-    public boolean isProxy() {
-        return proxyToTargetBrokerAddress != null;
-    }
-
-    public SocketAddress getLocalAddress() {
-        return this.localAddress;
-    }
+    private static final Logger log = LoggerFactory.getLogger(ClientCnx.class);
 
     /**
      * Check client connection is now free. This method will not change the state to idle.

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -124,7 +124,6 @@
 import org.apache.pulsar.common.api.proto.SingleMessageMetadata;
 import org.apache.pulsar.common.compression.CompressionCodec;
 import org.apache.pulsar.common.compression.CompressionCodecProvider;
-import org.apache.pulsar.common.naming.Metadata;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.common.schema.SchemaInfo;
@@ -191,7 +190,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
 
     private final MessageCrypto msgCrypto;
 
-    private Map<String, String> metadata;
+    private final Map<String, String> metadata;
 
     private final boolean readCompacted;
     private final boolean resetIncludeHead;
@@ -365,7 +364,7 @@ protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurat
         if (conf.getProperties().isEmpty()) {
             metadata = Collections.emptyMap();
         } else {
-            metadata = new HashMap<>(conf.getProperties());
+            metadata = Collections.unmodifiableMap(new HashMap<>(conf.getProperties()));
         }
 
         this.connectionHandler = new ConnectionHandler(this,
@@ -919,7 +918,6 @@ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
         // synchronized this, because redeliverUnAckMessage eliminate the epoch inconsistency between them
         final CompletableFuture<Void> future = new CompletableFuture<>();
         synchronized (this) {
-            updateProxyMetadataIfNeeded(cnx);
             ByteBuf request = Commands.newSubscribe(topic, subscription, consumerId, requestId, getSubType(),
                     priorityLevel, consumerName, isDurable, startMessageIdData, metadata, readCompacted,
                     conf.getReplicateSubscriptionState(),
@@ -3159,14 +3157,6 @@ private static BaseCommand newMultiMessageAckCommon(List<Triple<Long, Long, Conc
         return cmd;
     }
 
-    private void updateProxyMetadataIfNeeded(ClientCnx cnx) {
-        boolean isProxy = cnx.isProxy() || client.getConfiguration().getProxyServiceUrl() != null;
-        if (isProxy && cnx.getLocalAddress() != null) {
-            metadata = metadata.isEmpty() ? new HashMap<>() : metadata;
-            metadata.put(Metadata.CLIENT_IP, cnx.getLocalAddress().toString());
-        }
-    }
-
     private CompletableFuture<Void> doTransactionAcknowledgeForResponse(List<MessageId> messageIds, AckType ackType,
                                                                         Map<String, Long> properties, TxnID txnID) {
         long requestId = client.newRequestId();

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -92,7 +92,6 @@
 import org.apache.pulsar.common.api.proto.ProtocolVersion;
 import org.apache.pulsar.common.compression.CompressionCodec;
 import org.apache.pulsar.common.compression.CompressionCodecProvider;
-import org.apache.pulsar.common.naming.Metadata;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.protocol.ByteBufPair;
 import org.apache.pulsar.common.protocol.Commands;
@@ -157,7 +156,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
 
     private ScheduledFuture<?> keyGeneratorTask = null;
 
-    private Map<String, String> metadata;
+    private final Map<String, String> metadata;
 
     private Optional<byte[]> schemaVersion = Optional.empty();
 
@@ -281,7 +280,7 @@ public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfiguration
         if (conf.getProperties().isEmpty()) {
             metadata = Collections.emptyMap();
         } else {
-            metadata = new HashMap<>(conf.getProperties());
+            metadata = Collections.unmodifiableMap(new HashMap<>(conf.getProperties()));
         }
 
         InstrumentProvider ip = client.instrumentProvider();
@@ -1855,7 +1854,6 @@ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
         }
 
         final CompletableFuture<Void> future = new CompletableFuture<>();
-        updateProxyMetadataIfNeeded(cnx);
         cnx.sendRequestWithId(
                 Commands.newProducer(topic, producerId, requestId, producerName, conf.isEncryptionEnabled(), metadata,
                         schemaInfo, epoch, userProvidedProducerName,
@@ -2028,14 +2026,6 @@ public void connectionFailed(PulsarClientException exception) {
         }
     }
 
-    private void updateProxyMetadataIfNeeded(ClientCnx cnx) {
-        boolean isProxy = cnx.isProxy() || client.getConfiguration().getProxyServiceUrl() != null;
-        if (isProxy && cnx.getLocalAddress() != null) {
-            metadata = metadata.isEmpty() ? new HashMap<>() : metadata;
-            metadata.put(Metadata.CLIENT_IP, cnx.getLocalAddress().toString());
-        }
-    }
-
     private void closeProducerTasks() {
         Timeout timeout = sendTimeout;
         if (timeout != null) {

pulsar-common/src/main/java/org/apache/pulsar/common/naming/Metadata.java

@@ -25,7 +25,6 @@
  */
 public class Metadata {
 
-    public static final String CLIENT_IP = "X-Pulsar-Client-IP";
     private Metadata() {}
 
     public static void validateMetadata(Map<String, String> metadata,

pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyEnableHAProxyProtocolTest.java

@@ -122,11 +122,11 @@ public void testSimpleProduceAndConsume() throws PulsarClientException, PulsarAd
         SubscriptionStats subscriptionStats = topicStats.getSubscriptions().get(subName);
         Assert.assertEquals(subscriptionStats.getConsumers().size(), 1);
         Assert.assertEquals(subscriptionStats.getConsumers().get(0).getAddress(),
-                ((ConsumerImpl) consumer).getClientCnx().ctx().channel().localAddress().toString());
+                ((ConsumerImpl) consumer).getClientCnx().ctx().channel().localAddress().toString().replaceFirst("/", ""));
 
         topicStats = admin.topics().getStats(topicName);
         Assert.assertEquals(topicStats.getPublishers().size(), 1);
         Assert.assertEquals(topicStats.getPublishers().get(0).getAddress(),
-                ((ProducerImpl) producer).getClientCnx().ctx().channel().localAddress().toString());
+                ((ProducerImpl) producer).getClientCnx().ctx().channel().localAddress().toString().replaceFirst("/", ""));
     }
 }