Remove Custom...Server.java classes and replace with Thrift implementation

server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java

@@ -190,7 +190,10 @@ private static ServerAddress createThreadedSelectorServer(HostAndPort address,
       address = HostAndPort.fromParts(address.getHost(), transport.getPort());
     }
 
-    return new ServerAddress(new CustomThreadedSelectorServer(options), address);
+    final TThreadedSelectorServer server = new TThreadedSelectorServer(options);
+    server.setServerEventHandler(new ThriftServerEventHandler());
+
+    return new ServerAddress(server, address);
   }
 
   /**
@@ -225,7 +228,10 @@ private static ServerAddress createNonBlockingServer(HostAndPort address, TProce
       address = HostAndPort.fromParts(address.getHost(), transport.getPort());
     }
 
-    return new ServerAddress(new CustomNonBlockingServer(options), address);
+    CustomNonBlockingServer server = new CustomNonBlockingServer(options);
+    server.setServerEventHandler(new ThriftServerEventHandler());
+
+    return new ServerAddress(server, address);
   }
 
   /**
@@ -295,6 +301,8 @@ private static ServerAddress createBlockingServer(HostAndPort address, TProcesso
       log.info("Blocking Server bound on {}", address);
     }
 
+    server.setServerEventHandler(new ThriftServerEventHandler());
+
     return new ServerAddress(server, address);
 
   }
@@ -317,7 +325,11 @@ private static TThreadPoolServer createTThreadPoolServer(TServerTransport transp
     if (service != null) {
       options.executorService(service);
     }
-    return new TThreadPoolServer(options);
+
+    final TThreadPoolServer server = new TThreadPoolServer(options);
+    server.setServerEventHandler(new ThriftServerEventHandler());
+
+    return server;
   }
 
   /**
@@ -395,9 +407,11 @@ private static ServerAddress createSslThreadPoolServer(HostAndPort address, TPro
 
     ThreadPoolExecutor pool =
         createSelfResizingThreadPool(numThreads, threadTimeOut, conf, timeBetweenThreadChecks);
+    TThreadPoolServer server = createTThreadPoolServer(transport, processor,
+        ThriftUtil.transportFactory(), protocolFactory, pool);
+    server.setServerEventHandler(new ThriftServerEventHandler());
 
-    return new ServerAddress(createTThreadPoolServer(transport, processor,
-        ThriftUtil.transportFactory(), protocolFactory, pool), address);
+    return new ServerAddress(server, address);
   }
 
   private static ServerAddress createSaslThreadPoolServer(HostAndPort address, TProcessor processor,
@@ -495,6 +509,7 @@ private static ServerAddress createSaslThreadPoolServer(HostAndPort address, TPr
     final TThreadPoolServer server =
         createTThreadPoolServer(transport, processor, ugiTransportFactory, protocolFactory, pool);
 
+    server.setServerEventHandler(new ThriftServerEventHandler());
     return new ServerAddress(server, address);
   }
 

server/base/src/main/java/org/apache/accumulo/server/rpc/ThriftServerEventHandler.java

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.server.rpc;
+
+import java.net.SocketAddress;
+
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.server.ServerContext;
+import org.apache.thrift.server.TServerEventHandler;
+import org.apache.thrift.transport.TTransport;
+
+public class ThriftServerEventHandler implements TServerEventHandler {
+
+  public static class ThriftServerContext implements ServerContext {
+
+    @Override
+    public <T> T unwrap(Class<T> iface) {
+      return null;
+    }
+
+    @Override
+    public boolean isWrapperFor(Class<?> iface) {
+      return false;
+    }
+
+    @Override
+    public void setRemoteAddress(SocketAddress remoteAddress) {
+      TServerUtils.clientAddress.set(remoteAddress.toString());
+    }
+
+  }
+
+  @Override
+  public void preServe() {}
+
+  @Override
+  public ServerContext createContext(TProtocol input, TProtocol output) {
+    return new ThriftServerContext();
+  }
+
+  @Override
+  public void deleteContext(ServerContext serverContext, TProtocol input, TProtocol output) {
+    TServerUtils.clientAddress.set("");
+  }
+
+  @Override
+  public void processContext(ServerContext serverContext, TTransport inputTransport,
+      TTransport outputTransport) {}
+
+}