diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/BrokerProxyValidator.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/BrokerProxyValidator.java index debe1f7fcac87..b0529c2a777e1 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/BrokerProxyValidator.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/BrokerProxyValidator.java @@ -113,7 +113,7 @@ private static List parseCommaSeparatedConfigValue(String configValue) { } public CompletableFuture resolveAndCheckTargetAddress(String hostAndPort) { - int pos = hostAndPort.indexOf(':'); + int pos = hostAndPort.lastIndexOf(':'); String host = hostAndPort.substring(0, pos); int port = Integer.parseInt(hostAndPort.substring(pos + 1)); if (!isPortAllowed(port)) { diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java index edf4f18411366..0cbceb191b18d 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java @@ -26,9 +26,9 @@ import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelId; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.socket.SocketChannel; @@ -40,14 +40,8 @@ import io.netty.handler.ssl.SslHandler; import io.netty.handler.timeout.ReadTimeoutHandler; import io.netty.util.CharsetUtil; -import io.netty.util.concurrent.Future; -import io.netty.util.concurrent.FutureListener; import java.net.InetSocketAddress; -import java.net.URI; -import java.net.URISyntaxException; import java.util.Arrays; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; import javax.net.ssl.SSLSession; @@ -71,11 +65,11 @@ public class DirectProxyHandler { @Getter private final Channel inboundChannel; + private final ProxyConnection proxyConnection; @Getter Channel outboundChannel; @Getter private final Rate inboundChannelRequestsRate; - protected static Map inboundOutboundChannelMap = new ConcurrentHashMap<>(); private final String originalPrincipal; private final AuthData clientAuthData; private final String clientAuthMethod; @@ -86,12 +80,13 @@ public class DirectProxyHandler { private final ProxyService service; private final Runnable onHandshakeCompleteAction; - public DirectProxyHandler(ProxyService service, ProxyConnection proxyConnection, String targetBrokerUrl, + public DirectProxyHandler(ProxyService service, ProxyConnection proxyConnection, String brokerHostAndPort, InetSocketAddress targetBrokerAddress, int protocolVersion, Supplier sslHandlerSupplier) { this.service = service; this.authentication = proxyConnection.getClientAuthentication(); this.inboundChannel = proxyConnection.ctx().channel(); + this.proxyConnection = proxyConnection; this.inboundChannelRequestsRate = new Rate(); this.originalPrincipal = proxyConnection.clientAuthRole; this.clientAuthData = proxyConnection.clientAuthData; @@ -109,7 +104,18 @@ public DirectProxyHandler(ProxyService service, ProxyConnection proxyConnection, if (brokerProxyConnectTimeoutMs > 0) { b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, brokerProxyConnectTimeoutMs); } - b.group(inboundChannel.eventLoop()).channel(inboundChannel.getClass()).option(ChannelOption.AUTO_READ, false); + b.group(inboundChannel.eventLoop()) + .channel(inboundChannel.getClass()); + + String remoteHost; + try { + remoteHost = parseHost(brokerHostAndPort); + } catch (IllegalArgumentException e) { + log.warn("[{}] Failed to parse broker host '{}'", inboundChannel, brokerHostAndPort, e); + inboundChannel.close(); + return; + } + b.handler(new ChannelInitializer() { @Override protected void initChannel(SocketChannel ch) { @@ -123,55 +129,40 @@ protected void initChannel(SocketChannel ch) { } ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder( Commands.DEFAULT_MAX_MESSAGE_SIZE + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4)); - ch.pipeline().addLast("proxyOutboundHandler", new ProxyBackendHandler(config, protocolVersion)); + ch.pipeline().addLast("proxyOutboundHandler", + new ProxyBackendHandler(config, protocolVersion, remoteHost)); } }); - URI targetBroker; - try { - // targetBrokerUrl is coming in the "hostname:6650" form, so we need - // to extract host and port - targetBroker = new URI("pulsar://" + targetBrokerUrl); - } catch (URISyntaxException e) { - log.warn("[{}] Failed to parse broker url '{}'", inboundChannel, targetBrokerUrl, e); - inboundChannel.close(); - return; - } - ChannelFuture f = b.connect(targetBrokerAddress); outboundChannel = f.channel(); f.addListener(future -> { if (!future.isSuccess()) { // Close the connection if the connection attempt has failed. log.warn("[{}] Establishing connection to {} ({}) failed. Closing inbound channel.", inboundChannel, - targetBrokerAddress, targetBrokerUrl, future.cause()); + targetBrokerAddress, brokerHostAndPort, future.cause()); inboundChannel.close(); return; } - final ProxyBackendHandler cnx = (ProxyBackendHandler) outboundChannel.pipeline() - .get("proxyOutboundHandler"); - cnx.setRemoteHostName(targetBroker.getHost()); - - // if enable full parsing feature - if (service.getProxyLogLevel() == 2) { - //Set a map between inbound and outbound, - //so can find inbound by outbound or find outbound by inbound - inboundOutboundChannelMap.put(outboundChannel.id(), inboundChannel.id()); - } + }); + } - if (!config.isHaProxyProtocolEnabled()) { - return; - } + private static String parseHost(String brokerPortAndHost) { + int pos = brokerPortAndHost.lastIndexOf(':'); + if (pos > 0) { + return brokerPortAndHost.substring(0, pos); + } else { + throw new IllegalArgumentException("Illegal broker host:port '" + brokerPortAndHost + "'"); + } + } - if (proxyConnection.hasHAProxyMessage()) { - outboundChannel.writeAndFlush(encodeProxyProtocolMessage(proxyConnection.getHAProxyMessage())); - } else { - if (!(inboundChannel.remoteAddress() instanceof InetSocketAddress)) { - return; - } - if (!(outboundChannel.localAddress() instanceof InetSocketAddress)) { - return; - } + private void writeHAProxyMessage() { + if (proxyConnection.hasHAProxyMessage()) { + outboundChannel.writeAndFlush(encodeProxyProtocolMessage(proxyConnection.getHAProxyMessage())) + .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + } else { + if (inboundChannel.remoteAddress() instanceof InetSocketAddress + && outboundChannel.localAddress() instanceof InetSocketAddress) { InetSocketAddress clientAddress = (InetSocketAddress) inboundChannel.remoteAddress(); String sourceAddress = clientAddress.getAddress().getHostAddress(); int sourcePort = clientAddress.getPort(); @@ -179,13 +170,17 @@ protected void initChannel(SocketChannel ch) { String destinationAddress = proxyAddress.getAddress().getHostAddress(); int destinationPort = proxyAddress.getPort(); HAProxyMessage msg = new HAProxyMessage(HAProxyProtocolVersion.V1, HAProxyCommand.PROXY, - HAProxyProxiedProtocol.TCP4, sourceAddress, destinationAddress, sourcePort, destinationPort); - outboundChannel.writeAndFlush(encodeProxyProtocolMessage(msg)); + HAProxyProxiedProtocol.TCP4, sourceAddress, destinationAddress, sourcePort, + destinationPort); + outboundChannel.writeAndFlush(encodeProxyProtocolMessage(msg)) + .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); msg.release(); } - }); + } } + + private ByteBuf encodeProxyProtocolMessage(HAProxyMessage msg) { // Max length of v1 version proxy protocol message is 108 ByteBuf out = Unpooled.buffer(108); @@ -217,30 +212,45 @@ enum BackendState { Init, HandshakeCompleted } - public class ProxyBackendHandler extends PulsarDecoder implements FutureListener { + public class ProxyBackendHandler extends PulsarDecoder { private BackendState state = BackendState.Init; - private String remoteHostName; + private final String remoteHostName; protected ChannelHandlerContext ctx; private final ProxyConfiguration config; private final int protocolVersion; - public ProxyBackendHandler(ProxyConfiguration config, int protocolVersion) { + public ProxyBackendHandler(ProxyConfiguration config, int protocolVersion, String remoteHostName) { this.config = config; this.protocolVersion = protocolVersion; + this.remoteHostName = remoteHostName; } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { this.ctx = ctx; + + if (config.isHaProxyProtocolEnabled()) { + writeHAProxyMessage(); + } + // Send the Connect command to broker authenticationDataProvider = authentication.getAuthData(remoteHostName); AuthData authData = authenticationDataProvider.authenticate(AuthData.INIT_AUTH_DATA); ByteBuf command; command = Commands.newConnect(authentication.getAuthMethodName(), authData, protocolVersion, "Pulsar proxy", null /* target broker */, originalPrincipal, clientAuthData, clientAuthMethod); - outboundChannel.writeAndFlush(command); - outboundChannel.read(); + outboundChannel.writeAndFlush(command) + .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + } + + @Override + public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { + // handle backpressure + // stop/resume reading input from connection between the client and the proxy + // when the writability of the connection between the proxy and the broker changes + inboundChannel.config().setAutoRead(ctx.channel().isWritable()); + super.channelWritabilityChanged(ctx); } @Override @@ -261,7 +271,8 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exce if (msg instanceof ByteBuf) { ProxyService.BYTES_COUNTER.inc(((ByteBuf) msg).readableBytes()); } - inboundChannel.writeAndFlush(msg).addListener(this); + inboundChannel.writeAndFlush(msg) + .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); break; default: @@ -300,26 +311,13 @@ protected void handleAuthChallenge(CommandAuthChallenge authChallenge) { log.debug("{} Mutual auth {}", ctx.channel(), authentication.getAuthMethodName()); } - outboundChannel.writeAndFlush(request); - outboundChannel.read(); + outboundChannel.writeAndFlush(request) + .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); } catch (Exception e) { log.error("Error mutual verify", e); } } - @Override - public void operationComplete(Future future) { - // This is invoked when the write operation on the paired connection - // is completed - if (future.isSuccess()) { - outboundChannel.read(); - } else { - log.warn("[{}] [{}] Failed to write on proxy connection. Closing both connections.", inboundChannel, - outboundChannel, future.cause()); - inboundChannel.close(); - } - } - @Override protected void messageReceived() { // no-op @@ -349,18 +347,7 @@ protected void handleConnected(CommandConnected connected) { int maxMessageSize = connected.hasMaxMessageSize() ? connected.getMaxMessageSize() : Commands.INVALID_MAX_MESSAGE_SIZE; inboundChannel.writeAndFlush(Commands.newConnected(connected.getProtocolVersion(), maxMessageSize)) - .addListener(future -> { - if (future.isSuccess()) { - // Start reading from both connections - inboundChannel.read(); - outboundChannel.read(); - } else { - log.warn("[{}] [{}] Failed to write to inbound connection. Closing both connections.", - inboundChannel, - outboundChannel, future.cause()); - inboundChannel.close(); - } - }); + .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); } private void startDirectProxying(CommandConnected connected) { @@ -389,20 +376,20 @@ private void startDirectProxying(CommandConnected connected) { inboundChannel.pipeline().addBefore("handler", "inboundParser", new ParserProxyHandler(service, inboundChannel, ParserProxyHandler.FRONTEND_CONN, - connected.getMaxMessageSize())); + connected.getMaxMessageSize(), outboundChannel.id())); outboundChannel.pipeline().addBefore("proxyOutboundHandler", "outboundParser", new ParserProxyHandler(service, outboundChannel, ParserProxyHandler.BACKEND_CONN, - connected.getMaxMessageSize())); + connected.getMaxMessageSize(), inboundChannel.id())); } else { inboundChannel.pipeline().addBefore("handler", "inboundParser", new ParserProxyHandler(service, inboundChannel, ParserProxyHandler.FRONTEND_CONN, - Commands.DEFAULT_MAX_MESSAGE_SIZE)); + Commands.DEFAULT_MAX_MESSAGE_SIZE, outboundChannel.id())); outboundChannel.pipeline().addBefore("proxyOutboundHandler", "outboundParser", new ParserProxyHandler(service, outboundChannel, ParserProxyHandler.BACKEND_CONN, - Commands.DEFAULT_MAX_MESSAGE_SIZE)); + Commands.DEFAULT_MAX_MESSAGE_SIZE, inboundChannel.id())); } } } @@ -418,10 +405,6 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { ctx.close(); } - public void setRemoteHostName(String remoteHostName) { - this.remoteHostName = remoteHostName; - } - private boolean verifyTlsHostName(String hostname, ChannelHandlerContext ctx) { ChannelHandler sslHandler = ctx.channel().pipeline().get("tls"); diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java index fea1a401af61d..41a9b594f9363 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java @@ -25,6 +25,7 @@ import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelId; import io.netty.channel.ChannelInboundHandlerAdapter; import java.nio.charset.StandardCharsets; import java.util.ArrayList; @@ -53,6 +54,7 @@ public class ParserProxyHandler extends ChannelInboundHandlerAdapter { private final String connType; private final int maxMessageSize; + private final ChannelId peerChannelId; private final ProxyService service; @@ -66,11 +68,13 @@ public class ParserProxyHandler extends ChannelInboundHandlerAdapter { */ private static final Map consumerHashMap = new ConcurrentHashMap<>(); - public ParserProxyHandler(ProxyService service, Channel channel, String type, int maxMessageSize) { + public ParserProxyHandler(ProxyService service, Channel channel, String type, int maxMessageSize, + ChannelId peerChannelId) { this.service = service; this.channel = channel; this.connType = type; this.maxMessageSize = maxMessageSize; + this.peerChannelId = peerChannelId; } private void logging(Channel conn, BaseCommand.Type cmdtype, String info, List messages) { @@ -154,7 +158,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) { break; } topicName = TopicName.get(ParserProxyHandler.consumerHashMap.get(cmd.getMessage().getConsumerId() - + "," + DirectProxyHandler.inboundOutboundChannelMap.get(ctx.channel().id()))); + + "," + peerChannelId)); msgBytes = new MutableLong(0); MessageParser.parseMessage(topicName, -1L, -1L, buffer, (message) -> { diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 382e4751cd7c5..58203eee51c28 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -20,12 +20,11 @@ import static com.google.common.base.Preconditions.checkArgument; import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.haproxy.HAProxyMessage; import io.netty.handler.ssl.SslHandler; -import io.netty.util.concurrent.Future; -import io.netty.util.concurrent.FutureListener; import java.net.SocketAddress; import java.util.Collections; import java.util.List; @@ -65,7 +64,7 @@ * Handles incoming discovery request from client and sends appropriate response back to client. * */ -public class ProxyConnection extends PulsarHandler implements FutureListener { +public class ProxyConnection extends PulsarHandler { private static final Logger LOG = LoggerFactory.getLogger(ProxyConnection.class); // ConnectionPool is used by the proxy to issue lookup requests private ConnectionPool connectionPool; @@ -186,6 +185,17 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E ctx.close(); } + @Override + public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { + if (directProxyHandler != null && directProxyHandler.outboundChannel != null) { + // handle backpressure + // stop/resume reading input from connection between the proxy and the broker + // when the writability of the connection between the client and the proxy changes + directProxyHandler.outboundChannel.config().setAutoRead(ctx.channel().isWritable()); + } + super.channelWritabilityChanged(ctx); + } + @Override public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception { if (msg instanceof HAProxyMessage) { @@ -209,7 +219,8 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exce directProxyHandler.getInboundChannelRequestsRate().recordEvent(bytes); ProxyService.BYTES_COUNTER.inc(bytes); } - directProxyHandler.outboundChannel.writeAndFlush(msg).addListener(this); + directProxyHandler.outboundChannel.writeAndFlush(msg) + .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); break; default: @@ -217,18 +228,6 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exce } } - @Override - public void operationComplete(Future future) { - // This is invoked when the write operation on the paired connection is - // completed - if (future.isSuccess()) { - ctx.read(); - } else { - LOG.warn("[{}] Error in writing to inbound channel. Closing", remoteAddress, future.cause()); - directProxyHandler.outboundChannel.close(); - } - } - private synchronized void completeConnect(AuthData clientData) throws PulsarClientException { if (service.getConfiguration().isAuthenticationEnabled()) { if (service.getConfiguration().isForwardAuthorizationCredentials()) { @@ -266,18 +265,18 @@ private synchronized void completeConnect(AuthData clientData) throws PulsarClie ctx() .writeAndFlush( Commands.newError(-1, ServerError.ServiceNotReady, "Target broker isn't available.")) - .addListener(future -> ctx().close()); + .addListener(ChannelFutureListener.CLOSE); return; } brokerProxyValidator.resolveAndCheckTargetAddress(proxyToBrokerUrl) - .thenAccept(address -> ctx().executor().submit(() -> { + .thenAcceptAsync(address -> { // Client already knows which broker to connect. Let's open a // connection there and just pass bytes in both directions state = State.ProxyConnectionToBroker; directProxyHandler = new DirectProxyHandler(service, this, proxyToBrokerUrl, address, protocolVersionToAdvertise, sslHandlerSupplier); - })) + }, ctx.executor()) .exceptionally(throwable -> { if (throwable instanceof TargetAddressDeniedException || throwable.getCause() instanceof TargetAddressDeniedException) { @@ -296,7 +295,7 @@ private synchronized void completeConnect(AuthData clientData) throws PulsarClie .writeAndFlush( Commands.newError(-1, ServerError.ServiceNotReady, "Target broker cannot be validated.")) - .addListener(future -> ctx().close()); + .addListener(ChannelFutureListener.CLOSE); return null; }); } else { @@ -305,7 +304,8 @@ private synchronized void completeConnect(AuthData clientData) throws PulsarClie // partitions metadata lookups state = State.ProxyLookupRequests; lookupProxyHandler = new LookupProxyHandler(service, this); - ctx.writeAndFlush(Commands.newConnected(protocolVersionToAdvertise)); + ctx.writeAndFlush(Commands.newConnected(protocolVersionToAdvertise)) + .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); } } @@ -324,7 +324,8 @@ private void doAuthentication(AuthData clientData) throws Exception { } // auth not complete, continue auth with client side. - ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, protocolVersionToAdvertise)); + ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, protocolVersionToAdvertise)) + .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); if (LOG.isDebugEnabled()) { LOG.debug("[{}] Authentication in progress client by method {}.", remoteAddress, authMethod); @@ -402,8 +403,8 @@ remoteAddress, protocolVersionToAdvertise, getRemoteEndpointProtocolVersion(), doAuthentication(clientData); } catch (Exception e) { LOG.warn("[{}] Unable to authenticate: ", remoteAddress, e); - ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, "Failed to authenticate")); - close(); + ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, "Failed to authenticate")) + .addListener(ChannelFutureListener.CLOSE); } } @@ -424,8 +425,8 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) { } catch (Exception e) { String msg = "Unable to handleAuthResponse"; LOG.warn("[{}] {} ", remoteAddress, msg, e); - ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, msg)); - close(); + ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, msg)) + .addListener(ChannelFutureListener.CLOSE); } } @@ -458,25 +459,6 @@ protected void handleLookup(CommandLookupTopic lookup) { lookupProxyHandler.handleLookup(lookup); } - private synchronized void close() { - if (state != State.Closed) { - state = State.Closed; - if (directProxyHandler != null && directProxyHandler.outboundChannel != null) { - directProxyHandler.outboundChannel.close(); - directProxyHandler = null; - } - if (connectionPool != null) { - try { - connectionPool.close(); - connectionPool = null; - } catch (Exception e) { - LOG.error("Error closing connection pool", e); - } - } - ctx.close(); - } - } - ClientConfigurationData createClientConfiguration() { ClientConfigurationData clientConf = new ClientConfigurationData(); clientConf.setServiceUrl(service.getServiceUrl()); diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/BrokerProxyValidatorTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/BrokerProxyValidatorTest.java index 8e457554cf5ad..fba3c36e26616 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/BrokerProxyValidatorTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/BrokerProxyValidatorTest.java @@ -90,6 +90,26 @@ public void shouldAllowAllWithWildcard() throws Exception { brokerProxyValidator.resolveAndCheckTargetAddress("myhost.mydomain:6650").get(); } + @Test + public void shouldAllowIPv6Address() throws Exception { + BrokerProxyValidator brokerProxyValidator = new BrokerProxyValidator( + createMockedAddressResolver("fd4d:801b:73fa:abcd:0000:0000:0000:0001"), + "*" + , "fd4d:801b:73fa:abcd::/64" + , "6650"); + brokerProxyValidator.resolveAndCheckTargetAddress("myhost.mydomain:6650").get(); + } + + @Test + public void shouldAllowIPv6AddressNumeric() throws Exception { + BrokerProxyValidator brokerProxyValidator = new BrokerProxyValidator( + createMockedAddressResolver("fd4d:801b:73fa:abcd:0000:0000:0000:0001"), + "*" + , "fd4d:801b:73fa:abcd::/64" + , "6650"); + brokerProxyValidator.resolveAndCheckTargetAddress("fd4d:801b:73fa:abcd:0000:0000:0000:0001:6650").get(); + } + private AddressResolver createMockedAddressResolver(String ipAddressResult) { AddressResolver inetSocketAddressResolver = mock(AddressResolver.class); when(inetSocketAddressResolver.resolve(any())).then(invocationOnMock -> {