dcache-xroot: handle outbound errors on channel promise

Motivation:

Refer to the final comments in
GH 6909 "Proxy xrootd door keep restarting"
#6909

Occasionally we will find unexpected errors on the
pipeline being reported as with the example below

```
04 Dec 2022 04:07:55 (Xrootd-dcdndoor03-externalsubnet) [] 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.channel.StacklessClosedChannelException: null
	at io.netty.channel.AbstractChannel$AbstractUnsafe.write(Object, ChannelPromise)(Unknown Source)
```

The problem arises from the asymmetical nature of
exception propagation on the Netty pipeline.  For
exceptions on inbound/read, all that is needed is
for the last handler in the pipeline (the "TOP")
to implement exceptionCaught.  This is done by
the final handlers in the door, proxy and pool
pipelines.

However, for outbound exceptions, one needs to
add a listener to the channel future or channel
promise; this needs to be done at the very
beginning of the outbound pipeline ("BOTTOM");
the listener allows exceptions on channel write
to invoke the uncaughtException method on the
inbound handlers.

Modification:

Create an OutboundHandler that simply adds the
listener and add it first to all pipelines.

Result:

We should no longer see exceptions such as the one
above reporting no exception handler to handle
the exception.

Target: master
Request: 8.2
Request: 8.1 (Without proxy)
Request: 8.0 (Without proxy)
Request: 7.2 (Without proxy)
Closes: #6909
Patch: https://rb.dcache.org/r/13833/
Requires-notes: yes
Acked-by: Tigran

modules/dcache-xrootd/src/main/java/org/dcache/xrootd/OutboundExceptionHandler.java

@@ -0,0 +1,36 @@
+/* dCache - http://www.dcache.org/
+ *
+ * Copyright (C) 2014 Deutsches Elektronen-Synchrotron
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.dcache.xrootd;
+
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+
+/**
+ *  Should be added at the beginning of each pipeline to ensure that outbound errors
+ *  on the channel promise are handled.
+ */
+public class OutboundExceptionHandler extends ChannelOutboundHandlerAdapter {
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
+          throws Exception {
+        promise.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+        super.write(ctx, msg, promise);
+    }
+}

modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/NettyXrootdServer.java

@@ -35,6 +35,7 @@
 import java.util.concurrent.TimeUnit;
 import org.dcache.util.CDCThreadFactory;
 import org.dcache.util.NDC;
+import org.dcache.xrootd.OutboundExceptionHandler;
 import org.dcache.xrootd.core.XrootdAuthenticationHandler;
 import org.dcache.xrootd.core.XrootdDecoder;
 import org.dcache.xrootd.core.XrootdEncoder;
@@ -211,6 +212,7 @@ protected void initChannel(Channel ch) throws Exception {
                             Longs.toByteArray(sessionCounter.next()));
 
                       ChannelPipeline pipeline = ch.pipeline();
+                      pipeline.addLast("outerrors", new OutboundExceptionHandler());
                       pipeline.addLast("session", new SessionHandler(session));
                       if (_expectProxyProtocol) {
                           pipeline.addLast("haproxy", new HAProxyMessageDecoder());

modules/dcache-xrootd/src/main/java/org/dcache/xrootd/pool/XrootdTransferService.java

@@ -67,6 +67,7 @@
 import org.dcache.util.NetworkUtils;
 import org.dcache.vehicles.XrootdDoorAdressInfoMessage;
 import org.dcache.vehicles.XrootdProtocolInfo;
+import org.dcache.xrootd.OutboundExceptionHandler;
 import org.dcache.xrootd.core.XrootdDecoder;
 import org.dcache.xrootd.core.XrootdEncoder;
 import org.dcache.xrootd.core.XrootdHandshakeHandler;
@@ -397,7 +398,7 @@ protected void initChannel(Channel ch) throws Exception {
         super.initChannel(ch);
 
         ChannelPipeline pipeline = ch.pipeline();
-
+        pipeline.addLast("outerrors", new OutboundExceptionHandler());
         pipeline.addLast("handshake",
               new XrootdHandshakeHandler(XrootdProtocol.DATA_SERVER));
         pipeline.addLast("encoder", new XrootdEncoder());