Repository: kroxylicious/kroxylicious (java)

Issue:
Poor error reporting if SNI name doesn't match a bootstrap/broker hostname defined by a virtualcluster
**Describe the bug**

If you mis-configure SNI, the error message in the logs is not too helpful:

```
No channel bindings found for binding address: <any>, port: 9092, sniHostname: <none>, tls: true
```

The exception message doesn't mention that SNI hostname, which makes it tricky to diagnose.  Netty invokes the SNI handler twice, the first time with the SNI name, then a second time with a null SNI.  It throws the exception if it fails on the 2nd attempt.

```
2025-02-11 22:40:36 <epollEventLoopGroup-5-2> WARN  io.netty.channel.DefaultChannelPipeline - An exceptionCaught() event was fired, and it reached
at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.io.netty.handler.codec.DecoderException: failed to get the SslContext for null
	at io.netty.handler.ssl.SniHandler.onLookupComplete(SniHandler.java:158) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.ssl.AbstractSniHandler.onLookupComplete(AbstractSniHandler.java:191) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]
	at io.kroxylicious.proxy.internal.KafkaProxyInitializer$2.onLookupComplete(KafkaProxyInitializer.java:161) ~[kroxylicious-runtime-0.10.0.jar:?]
	at io.netty.handler.ssl.SslClientHelloHandler.select(SslClientHelloHandler.java:222) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.ssl.SslClientHelloHandler.decode(SslClientHelloHandler.java:201) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:530) ~[netty-codec-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:469) ~[netty-codec-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:427) ~[netty-codec-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:394) ~[netty-codec-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:377) ~[netty-codec-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303) ~[netty-transport-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281) ~[netty-transport-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274) ~[netty-transport-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1352) ~[netty-transport-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:301) ~[netty-transport-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281) ~[netty-transport-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:850) ~[netty-transport-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:811) ~[netty-transport-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173) ~[netty-common-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166) ~[netty-common-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472) ~[netty-common-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:405) ~[netty-transport-classes-epoll-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[netty-common-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.117.Final.jar:4.1.117.Final]
	at java.base/java.lang.Thread.run(Thread.java:840) [?:?]
Caused by: java.util.concurrent.CompletionException: io.kroxylicious.proxy.internal.net.EndpointResolutionException: No channel bindings found for binding address: <any>, port: 9092, sniHostname: <none>, tls: true
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:687) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:662) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2168) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:144) ~[?:?]
	at io.kroxylicious.proxy.internal.net.EndpointRegistry.resolve(EndpointRegistry.java:542) ~[kroxylicious-runtime-0.10.0.jar:?]
	at io.kroxylicious.proxy.internal.KafkaProxyInitializer.lambda$initTlsChannel$1(KafkaProxyInitializer.java:129) ~[kroxylicious-runtime-0.10.0.jar:?]
	at io.netty.handler.ssl.SniHandler.lookup(SniHandler.java:147) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.ssl.AbstractSniHandler.lookup(AbstractSniHandler.java:182) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.ssl.SslClientHelloHandler.select(SslClientHelloHandler.java:220) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]	... 22 more
Caused by: io.kroxylicious.proxy.internal.net.EndpointResolutionException: No channel bindings found for binding address: <any>, port: 9092, sniHostname: <none>, tls: true
	at io.kroxylicious.proxy.internal.net.EndpointRegistry.buildEndpointResolutionException(EndpointRegistry.java:601) ~[kroxylicious-runtime-0.10.0.jar:?]
	at io.kroxylicious.proxy.internal.net.EndpointRegistry.lambda$resolve$30(EndpointRegistry.java:564) ~[kroxylicious-runtime-0.10.0.jar:?]
	at java.base/java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:684) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:662) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2168) ~[?:?]
	at java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:144) ~[?:?]
	at io.kroxylicious.proxy.internal.net.EndpointRegistry.resolve(EndpointRegistry.java:542) ~[kroxylicious-runtime-0.10.0.jar:?]
	at io.kroxylicious.proxy.internal.KafkaProxyInitializer.lambda$initTlsChannel$1(KafkaProxyInitializer.java:129) ~[kroxylicious-runtime-0.10.0.jar:?]
	at io.netty.handler.ssl.SniHandler.lookup(SniHandler.java:147) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.ssl.AbstractSniHandler.lookup(AbstractSniHandler.java:182) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]
	at io.netty.handler.ssl.SslClientHelloHandler.select(SslClientHelloHandler.java:220) ~[netty-handler-4.1.117.Final.jar:4.1.117.Final]	... 22 more
```


**To Reproduce**
Steps to reproduce the behavior:
1. Configure kroxylicious like this ...
2. Run command '....'
3. See error

**Expected behavior**
A clear and concise description of what you expected to happen.

**YAML files and logs**

Attach or copy paste your kroxylicious configuration YAML (with sensitive details
like broker address redacted).
Attach or copy and paste relevant logs.

**Additional context**
Add any other context about the problem here.


Pull request description:
Improve logging on SNI binding resolution failure
### Type of change

- Bugfix

### Description

Improve logging on SNI mismatch.

### Additional Context

Currently when the client sends an SNI that doesn't match the virtual cluster configuration a misleading exception is logged. The netty code retries with a null sniHostname and that is what you see in the logs, you don't see the initial sniHostname the client sent anywhere.

This change also installs and error handler at the end of the frontend pipeline so that we can control the log verbosity of any uncaught exceptions that percolate to the end of the pipeline.

Closes #1811

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [ ] PR raised from a fork of this repository and made from a branch rather than main. 
- [ ] Write tests
- [ ] Update documentation
- [ ] Make sure all unit/integration tests pass
- [ ] Make sure all Sonarcloud warnings are addressed or are justifiably ignored.
- [ ] If applicable to the change, [trigger the system test suite](../blob/main/DEV_GUIDE.md#jenkins-pipeline-for-system-tests).  Make sure tests pass.
- [ ] If applicable to the change, [trigger the performance test suite](../blob/main/PERFORMANCE.md#jenkins-pipeline-for-performance). Ensure that any degradations to performance numbers are understood and justified.
- [ ] Ensure the PR references relevant issue(s) so they are closed on merging.
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).

> **_NOTE:_**  You must be a member of `@kroxylicious/developers` to trigger the system test and performance test suites.  If you are not part of this group, comment on the PR requesting a trigger, tagging `@kroxylicious/developers`.


Patch:
diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
index 84d7fe66..868b9950 100644
--- a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
+++ b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyInitializer.java
@@ -49,6 +49,10 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProxyInitializer.class);
 
+    private static final ChannelInboundHandlerAdapter LOGGING_INBOUND_ERROR_HANDLER = new LoggingInboundErrorHandler();
+    @VisibleForTesting
+    static final String LOGGING_INBOUND_ERROR_HANDLER_NAME = "loggingInboundErrorHandler";
+
     private final boolean haproxyProtocol;
     private final Map<KafkaAuthnHandler.SaslMechanism, AuthenticateCallbackHandler> authnHandlers;
     private final boolean tls;
@@ -92,11 +96,12 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
         else {
             initPlainChannel(ch, pipeline, bindingAddress, targetPort);
         }
+        addLoggingErrorHandler(pipeline);
     }
 
     @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
     private void initPlainChannel(SocketChannel ch, ChannelPipeline pipeline, Optional<String> bindingAddress, int targetPort) {
-        pipeline.addLast(new ChannelInboundHandlerAdapter() {
+        pipeline.addLast("plainResolver", new ChannelInboundHandlerAdapter() {
             @Override
             public void channelActive(ChannelHandlerContext ctx) {
                 virtualClusterBindingResolver.resolve(Endpoint.createEndpoint(bindingAddress, targetPort, tls), null)
@@ -121,16 +126,19 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
         });
     }
 
-    @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+    // deep inheritance tree of SniHandler not something we can fix
+    @SuppressWarnings({ "OptionalUsedAsFieldOrParameterType", "java:S110" })
     private void initTlsChannel(SocketChannel ch, ChannelPipeline pipeline, Optional<String> bindingAddress, int targetPort) {
         LOGGER.debug("Adding SSL/SNI handler");
-        pipeline.addLast(new SniHandler((sniHostname, promise) -> {
+        pipeline.addLast("sniResolver", new SniHandler((sniHostname, promise) -> {
             try {
-                var stage = virtualClusterBindingResolver.resolve(Endpoint.createEndpoint(bindingAddress, targetPort, tls), sniHostname);
+                Endpoint endpoint = Endpoint.createEndpoint(bindingAddress, targetPort, tls);
+                var stage = virtualClusterBindingResolver.resolve(endpoint, sniHostname);
                 // completes the netty promise when then resolution completes (success/otherwise).
                 stage.handle((binding, t) -> {
                     try {
                         if (t != null) {
+                            LOGGER.warn("Exception resolving Virtual Cluster Binding for endpoint {} and sniHostname {}: {}", endpoint, sniHostname, t.getMessage());
                             promise.setFailure(t);
                             return null;
                         }
@@ -168,6 +176,7 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
     void addHandlers(SocketChannel ch, VirtualClusterBinding binding) {
         var virtualCluster = binding.virtualClusterModel();
         ChannelPipeline pipeline = ch.pipeline();
+        pipeline.remove(LOGGING_INBOUND_ERROR_HANDLER_NAME);
         if (virtualCluster.isLogNetwork()) {
             pipeline.addLast("networkLogger", new LoggingHandler("io.kroxylicious.proxy.internal.DownstreamNetworkLogger", LogLevel.INFO));
         }
@@ -207,10 +216,15 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
         var frontendHandler = new KafkaProxyFrontendHandler(netFilter, dp, virtualCluster);
 
         pipeline.addLast("netHandler", frontendHandler);
+        addLoggingErrorHandler(pipeline);
 
         LOGGER.debug("{}: Initial pipeline: {}", ch, pipeline);
     }
 
+    private static void addLoggingErrorHandler(ChannelPipeline pipeline) {
+        pipeline.addLast(LOGGING_INBOUND_ERROR_HANDLER_NAME, LOGGING_INBOUND_ERROR_HANDLER);
+    }
+
     @VisibleForTesting
     static class InitalizerNetFilter implements NetFilter {
 
@@ -271,4 +285,16 @@ public class KafkaProxyInitializer extends ChannelInitializer<SocketChannel> {
             context.initiateConnect(target, filters);
         }
     }
+
+    @Sharable
+    @VisibleForTesting
+    static class LoggingInboundErrorHandler extends ChannelInboundHandlerAdapter {
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+            LOGGER.atWarn()
+                    .setCause(LOGGER.isDebugEnabled() ? cause : null)
+                    .log("An exceptionCaught() event was caught by the error handler {}: {}. Increase log level to DEBUG for stacktrace",
+                            cause.getClass().getSimpleName(), cause.getMessage());
+        }
+    }
 }


Test patch:
diff --git a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyInitializerTest.java b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyInitializerTest.java
index f5b55122..dd54686e 100644
--- a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyInitializerTest.java
+++ b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyInitializerTest.java
@@ -23,12 +23,14 @@ import org.mockito.ArgumentCaptor;
 import org.mockito.Captor;
 import org.mockito.InOrder;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.junit.jupiter.MockitoExtension;
 
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.EventLoop;
+import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.ByteToMessageDecoder;
@@ -51,7 +53,12 @@ import io.kroxylicious.proxy.model.VirtualClusterModel;
 import io.kroxylicious.proxy.service.ClusterNetworkAddressConfigProvider;
 import io.kroxylicious.proxy.service.HostPort;
 
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+import static io.kroxylicious.proxy.internal.KafkaProxyInitializer.LOGGING_INBOUND_ERROR_HANDLER_NAME;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.ArgumentMatchers.isNull;
@@ -113,19 +120,14 @@ class KafkaProxyInitializerTest {
     @Test
     void shouldInitialisePlainChannel() {
         // Given
-        kafkaProxyInitializer = new KafkaProxyInitializer(filterChainFactory,
-                pfr,
-                false,
-                (endpoint, sniHostname) -> bindingStage,
-                (virtualCluster, upstreamNodes) -> null,
-                false,
-                Map.of(), new ApiVersionsServiceImpl());
+        kafkaProxyInitializer = createKafkaProxyInitializer(false, (endpoint, sniHostname) -> bindingStage, Map.of());
         // When
         kafkaProxyInitializer.initChannel(channel);
 
         // Then
-        verify(channelPipeline).addLast(isA(ChannelInboundHandlerAdapter.class));
-        verify(channelPipeline, times(0)).addLast(isA(SniHandler.class));
+        verify(channelPipeline).addLast(eq("plainResolver"), isA(ChannelInboundHandlerAdapter.class));
+        assertErrorHandlerAdded();
+        verify(channelPipeline, times(0)).addLast(anyString(), isA(SniHandler.class));
     }
 
     @Test
@@ -133,14 +135,8 @@ class KafkaProxyInitializerTest {
         // Given
         final VirtualClusterBindingResolver virtualClusterBindingResolver = mock(VirtualClusterBindingResolver.class);
         when(virtualClusterBindingResolver.resolve(any(Endpoint.class), isNull())).thenReturn(bindingStage);
-        kafkaProxyInitializer = new KafkaProxyInitializer(filterChainFactory,
-                pfr,
-                false,
-                virtualClusterBindingResolver,
-                (virtualCluster, upstreamNodes) -> null,
-                false,
-                Map.of(), new ApiVersionsServiceImpl());
-        when(channelPipeline.addLast(plainChannelResolverCaptor.capture())).thenReturn(channelPipeline);
+        kafkaProxyInitializer = createKafkaProxyInitializer(false, virtualClusterBindingResolver, Map.of());
+        when(channelPipeline.addLast(eq("plainResolver"), plainChannelResolverCaptor.capture())).thenReturn(channelPipeline);
 
         kafkaProxyInitializer.initChannel(channel);
         final ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
@@ -157,14 +153,8 @@ class KafkaProxyInitializerTest {
         // Given
         final VirtualClusterBindingResolver virtualClusterBindingResolver = mock(VirtualClusterBindingResolver.class);
         when(virtualClusterBindingResolver.resolve(any(Endpoint.class), isNull())).thenReturn(bindingStage);
-        kafkaProxyInitializer = new KafkaProxyInitializer(filterChainFactory,
-                pfr,
-                false,
-                virtualClusterBindingResolver,
-                (virtualCluster, upstreamNodes) -> null,
-                false,
-                Map.of(), new ApiVersionsServiceImpl());
-        when(channelPipeline.addLast(plainChannelResolverCaptor.capture())).thenReturn(channelPipeline);
+        kafkaProxyInitializer = createKafkaProxyInitializer(false, virtualClusterBindingResolver, Map.of());
+        when(channelPipeline.addLast(eq("plainResolver"), plainChannelResolverCaptor.capture())).thenReturn(channelPipeline);
 
         kafkaProxyInitializer.initChannel(channel);
         final ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
@@ -182,23 +172,18 @@ class KafkaProxyInitializerTest {
         // Given
         when(vcb.virtualClusterModel())
                 .thenReturn(virtualClusterModel);
-        kafkaProxyInitializer = new KafkaProxyInitializer(filterChainFactory,
-                pfr,
-                tls,
-                (endpoint, sniHostname) -> bindingStage,
-                (virtualCluster, upstreamNodes) -> null,
-                false,
-                Map.of(), new ApiVersionsServiceImpl());
+        kafkaProxyInitializer = createKafkaProxyInitializer(tls, (endpoint, sniHostname) -> bindingStage, Map.of());
 
         // When
         kafkaProxyInitializer.addHandlers(channel, vcb);
 
         // Then
-        final InOrder verifyer = inOrder(channelPipeline);
-        verifyer.verify(channelPipeline).addLast(eq("requestDecoder"), any(ByteToMessageDecoder.class));
-        verifyer.verify(channelPipeline).addLast(eq("responseEncoder"), any(MessageToByteEncoder.class));
-        verifyer.verify(channelPipeline).addLast(eq("responseOrderer"), any(ResponseOrderer.class));
-        verifyer.verify(channelPipeline).addLast(eq("netHandler"), any(KafkaProxyFrontendHandler.class));
+        final InOrder orderedVerifyer = inOrder(channelPipeline);
+        verifyErrorHandlerRemoved(orderedVerifyer);
+        verifyEncoderAndOrdererAdded(orderedVerifyer);
+        verifyFrontendHandlerAdded(orderedVerifyer);
+        verifyErrorHandlerAdded(orderedVerifyer);
+        Mockito.verifyNoMoreInteractions(channelPipeline);
     }
 
     @ParameterizedTest
@@ -207,19 +192,19 @@ class KafkaProxyInitializerTest {
         // Given
         virtualClusterModel = buildVirtualCluster(false, true);
         when(vcb.virtualClusterModel()).thenReturn(virtualClusterModel);
-        kafkaProxyInitializer = new KafkaProxyInitializer(filterChainFactory,
-                pfr,
-                tls,
-                (endpoint, sniHostname) -> bindingStage,
-                (virtualCluster, upstreamNodes) -> null,
-                false,
-                Map.of(), new ApiVersionsServiceImpl());
+        kafkaProxyInitializer = createKafkaProxyInitializer(tls, (endpoint, sniHostname) -> bindingStage, Map.of());
 
         // When
         kafkaProxyInitializer.addHandlers(channel, vcb);
 
         // Then
-        verify(channelPipeline).addLast(eq("frameLogger"), any(LoggingHandler.class));
+        final InOrder orderedVerifyer = inOrder(channelPipeline);
+        verifyErrorHandlerRemoved(orderedVerifyer);
+        verifyEncoderAndOrdererAdded(orderedVerifyer);
+        verifyFrameLoggerAdded(orderedVerifyer);
+        verifyFrontendHandlerAdded(orderedVerifyer);
+        verifyErrorHandlerAdded(orderedVerifyer);
+        Mockito.verifyNoMoreInteractions(channelPipeline);
     }
 
     @ParameterizedTest
@@ -228,62 +213,41 @@ class KafkaProxyInitializerTest {
         // Given
         virtualClusterModel = buildVirtualCluster(true, false);
         when(vcb.virtualClusterModel()).thenReturn(virtualClusterModel);
-        kafkaProxyInitializer = new KafkaProxyInitializer(filterChainFactory,
-                pfr,
-                tls,
-                (endpoint, sniHostname) -> bindingStage,
-                (virtualCluster, upstreamNodes) -> null,
-                false,
-                Map.of(), new ApiVersionsServiceImpl());
+        kafkaProxyInitializer = createKafkaProxyInitializer(tls, (endpoint, sniHostname) -> bindingStage, Map.of());
 
         // When
         kafkaProxyInitializer.addHandlers(channel, vcb);
 
         // Then
-        verify(channelPipeline).addLast(eq("networkLogger"), any(LoggingHandler.class));
+        final InOrder orderedVerifyer = inOrder(channelPipeline);
+        verifyErrorHandlerRemoved(orderedVerifyer);
+        verifyNetworkLoggerAdded(orderedVerifyer);
+        verifyEncoderAndOrdererAdded(orderedVerifyer);
+        verifyFrontendHandlerAdded(orderedVerifyer);
+        verifyErrorHandlerAdded(orderedVerifyer);
+        Mockito.verifyNoMoreInteractions(channelPipeline);
     }
 
     @ParameterizedTest
     @ValueSource(booleans = { true, false })
     void shouldAddAuthnHandlersOnBindingComplete(boolean tls) {
         // Given
-        virtualClusterModel = buildVirtualCluster(true, false);
+        virtualClusterModel = buildVirtualCluster(false, false);
         when(vcb.virtualClusterModel()).thenReturn(virtualClusterModel);
         final AuthenticateCallbackHandler plainHandler = mock(AuthenticateCallbackHandler.class);
-        kafkaProxyInitializer = new KafkaProxyInitializer(filterChainFactory,
-                pfr,
-                tls,
-                (endpoint, sniHostname) -> bindingStage,
-                (virtualCluster, upstreamNodes) -> null,
-                false,
-                Map.of(KafkaAuthnHandler.SaslMechanism.PLAIN, plainHandler), new ApiVersionsServiceImpl());
-
-        // When
-        kafkaProxyInitializer.addHandlers(channel, vcb);
-
-        // Then
-        verify(channelPipeline).addLast(any(KafkaAuthnHandler.class));
-    }
-
-    @ParameterizedTest
-    @ValueSource(booleans = { true, false })
-    void shouldNotAddAuthnHandlersWithoutConfiguredMechanism(boolean tls) {
-        // Given
-        virtualClusterModel = buildVirtualCluster(true, false);
-        when(vcb.virtualClusterModel()).thenReturn(virtualClusterModel);
-        kafkaProxyInitializer = new KafkaProxyInitializer(filterChainFactory,
-                pfr,
-                tls,
-                (endpoint, sniHostname) -> bindingStage,
-                (virtualCluster, upstreamNodes) -> null,
-                false,
-                Map.of(), new ApiVersionsServiceImpl());
+        kafkaProxyInitializer = createKafkaProxyInitializer(tls, (endpoint, sniHostname) -> bindingStage, Map.of(KafkaAuthnHandler.SaslMechanism.PLAIN, plainHandler));
 
         // When
         kafkaProxyInitializer.addHandlers(channel, vcb);
 
         // Then
-        verify(channelPipeline, times(0)).addLast(any(KafkaAuthnHandler.class));
+        final InOrder orderedVerifyer = inOrder(channelPipeline);
+        verifyErrorHandlerRemoved(orderedVerifyer);
+        verifyEncoderAndOrdererAdded(orderedVerifyer);
+        orderedVerifyer.verify(channelPipeline).addLast(any(KafkaAuthnHandler.class));
+        verifyFrontendHandlerAdded(orderedVerifyer);
+        verifyErrorHandlerAdded(orderedVerifyer);
+        Mockito.verifyNoMoreInteractions(channelPipeline);
     }
 
     @Test
@@ -313,18 +277,69 @@ class KafkaProxyInitializerTest {
     @Test
     void shouldInitialiseTlsChannel() {
         // Given
-        kafkaProxyInitializer = new KafkaProxyInitializer(filterChainFactory,
-                pfr,
-                true,
-                (endpoint, sniHostname) -> bindingStage,
-                (virtualCluster, upstreamNodes) -> null,
-                false,
-                Map.of(), new ApiVersionsServiceImpl());
+        kafkaProxyInitializer = createKafkaProxyInitializer(true, (endpoint, sniHostname) -> bindingStage, Map.of());
 
         // When
         kafkaProxyInitializer.initChannel(channel);
 
         // Then
-        verify(channelPipeline).addLast(isA(SniHandler.class));
+        verify(channelPipeline).addLast(anyString(), isA(SniHandler.class));
+        assertErrorHandlerAdded();
+    }
+
+    @Test
+    void testLoggingErrorHandlerPreventsExceptionPropagatingToChannel() {
+        EmbeddedChannel embeddedChannel = new EmbeddedChannel();
+        embeddedChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
+            @Override
+            public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                throw new RuntimeException("failed to handle message: " + msg);
+            }
+        });
+        embeddedChannel.pipeline().addLast(new KafkaProxyInitializer.LoggingInboundErrorHandler());
+        embeddedChannel.writeInbound("arbitrary");
+        assertThatCode(embeddedChannel::checkException).doesNotThrowAnyException();
+    }
+
+    private @NonNull KafkaProxyInitializer createKafkaProxyInitializer(boolean tls,
+                                                                       VirtualClusterBindingResolver virtualClusterBindingResolver,
+                                                                       Map<KafkaAuthnHandler.SaslMechanism, AuthenticateCallbackHandler> authnMechanismHandlers) {
+        return new KafkaProxyInitializer(filterChainFactory,
+                pfr,
+                tls,
+                virtualClusterBindingResolver,
+                (virtualCluster, upstreamNodes) -> null,
+                false,
+                authnMechanismHandlers, new ApiVersionsServiceImpl());
+    }
+
+    private void assertErrorHandlerAdded() {
+        verify(channelPipeline).addLast(anyString(), isA(KafkaProxyInitializer.LoggingInboundErrorHandler.class));
+    }
+
+    private void verifyFrameLoggerAdded(InOrder orderedVerifyer) {
+        orderedVerifyer.verify(channelPipeline).addLast(eq("frameLogger"), any(LoggingHandler.class));
+    }
+
+    private void verifyErrorHandlerAdded(InOrder orderedVerifyer) {
+        orderedVerifyer.verify(channelPipeline).addLast(eq(LOGGING_INBOUND_ERROR_HANDLER_NAME), isA(KafkaProxyInitializer.LoggingInboundErrorHandler.class));
+    }
+
+    private void verifyFrontendHandlerAdded(InOrder orderedVerifyer) {
+        orderedVerifyer.verify(channelPipeline).addLast(eq("netHandler"), any(KafkaProxyFrontendHandler.class));
+    }
+
+    private void verifyErrorHandlerRemoved(InOrder orderedVerifyer) {
+        orderedVerifyer.verify(channelPipeline).remove(LOGGING_INBOUND_ERROR_HANDLER_NAME);
+    }
+
+    private void verifyNetworkLoggerAdded(InOrder orderedVerifyer) {
+        orderedVerifyer.verify(channelPipeline).addLast(eq("networkLogger"), any(LoggingHandler.class));
+    }
+
+    private void verifyEncoderAndOrdererAdded(InOrder orderedVerifyer) {
+        orderedVerifyer.verify(channelPipeline).addLast(eq("requestDecoder"), any(ByteToMessageDecoder.class));
+        orderedVerifyer.verify(channelPipeline).addLast(eq("responseEncoder"), any(MessageToByteEncoder.class));
+        orderedVerifyer.verify(channelPipeline).addLast(eq("responseOrderer"), any(ResponseOrderer.class));
     }
 }