xrootd: Add session to access log

Motivation:

Netty processes all requests on a thread pool associated with the channel
context. When writing to a channel from another thread, the request is
first moved to the appropriate thread pool. In doing so the CDC is lost,
preventing the session to be logged to the access log.

Modification:

Wraps the response write in a runnable submitted on the appropriate
thread pool. Netty detects the call is from the correct thread and will
not resubmit the write itself. This gives us a chance to inject the
correct CDC.

Result:

The xrootd door access log logs the correct session. Fixes #1712.

Target: trunk
Request: 2.13
Request: 2.12
Require-notes: yes
Require-book: no
Acked-by: Paul Millar <paul.millar@desy.de>
Patch: https://rb.dcache.org/r/8370/
(cherry picked from commit 73441f8)

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

@@ -17,13 +17,17 @@
  */
 package org.dcache.xrootd.door;
 
-import com.google.common.collect.Sets;
 import com.google.common.net.InetAddresses;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.ChannelPromiseNotifier;
+import io.netty.util.concurrent.GenericFutureListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,6 +48,7 @@
 import diskCacheV111.util.PermissionDeniedCacheException;
 import diskCacheV111.util.TimeoutCacheException;
 
+import dmg.cells.nucleus.CDC;
 import dmg.cells.nucleus.CellPath;
 
 import org.dcache.auth.LoginReply;
@@ -111,6 +116,19 @@ public XrootdRedirectHandler(XrootdDoor door, FsPath rootPath, FsPath uploadPath
         _executor = executor;
     }
 
+    @Override
+    protected ChannelFuture respond(ChannelHandlerContext ctx, Object response)
+    {
+        CDC cdc = new CDC();
+        ChannelPromise promise = ctx.newPromise();
+        ctx.executor().execute(() -> {
+            try (CDC ignored = cdc.restore()) {
+                super.respond(ctx, response).addListener(new ChannelPromiseNotifier(promise));
+            }
+        });
+        return promise;
+    }
+
     @Override
     protected void requestReceived(ChannelHandlerContext ctx, XrootdRequest req)
     {