Address PR feedback.

Adds a check to ensure processors are only registered once.

Replaces duplicative RpcService usages with the getService() method.

Removes an unnecessary TMessage object creation.

Author: ddanielr

core/src/main/java/org/apache/accumulo/core/rpc/AccumuloTMultiplexedProcessor.java

@@ -29,6 +29,8 @@
 import org.apache.thrift.protocol.TProtocolDecorator;
 import org.apache.thrift.protocol.TProtocolException;
 
+import com.google.common.base.Preconditions;
+
 /**
  * AccumuloTMultiplexedProcessor is a {@link TMultiplexedProcessor} variant that uses the
  * {@link AccumuloTMultiplexedProtocol} to register additional processors and then forward messages
@@ -40,7 +42,11 @@ public class AccumuloTMultiplexedProcessor extends TMultiplexedProcessor {
   private final TProcessor[] PROCESSORS = new TProcessor[RpcService.values().length];
 
   public void registerProcessor(RpcService service, TProcessor processor) {
-    PROCESSORS[Byte.toUnsignedInt(service.getShortId())] = Objects.requireNonNull(processor,
+    int serviceID = Byte.toUnsignedInt(service.getShortId());
+    // Ensure that TProcessors are only registered once.
+    Preconditions.checkState(PROCESSORS[serviceID] == null,
+        "processor already exists with id " + serviceID);
+    PROCESSORS[serviceID] = Objects.requireNonNull(processor,
         "processor must not be null for RPCService: " + service.name());
   }
 

core/src/main/java/org/apache/accumulo/core/rpc/AccumuloTMultiplexedProtocol.java

@@ -49,7 +49,7 @@ public AccumuloTMultiplexedProtocol(TProtocol protocol, RpcService rpcService) {
   @Override
   public void writeMessageBegin(TMessage tMessage) throws TException {
     if (tMessage.type == TMessageType.CALL || tMessage.type == TMessageType.ONEWAY) {
-      super.writeMessageBegin(new TMessage(tMessage.name, tMessage.type, tMessage.seqid));
+      super.writeMessageBegin(tMessage);
       super.writeByte(this.rpcService.getShortId());
     } else {
       super.writeMessageBegin(tMessage);

core/src/main/java/org/apache/accumulo/core/rpc/clients/ClientServiceThriftClient.java

@@ -44,7 +44,7 @@ public class ClientServiceThriftClient extends ThriftClientTypes<Client>
   public Pair<String,Client> getThriftServerConnection(ClientContext context,
       boolean preferCachedConnections) throws TTransportException {
     return getThriftServerConnection(LOG, this, context, preferCachedConnections,
-        warnedAboutTServersBeingDown, RpcService.CLIENT);
+        warnedAboutTServersBeingDown, getService());
   }
 
   @Override

core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletManagementClientServiceThriftClient.java

@@ -49,7 +49,7 @@ public TabletManagementClientServiceThriftClient(RpcService service) {
   public Pair<String,Client> getThriftServerConnection(ClientContext context,
       boolean preferCachedConnections) throws TTransportException {
     return getThriftServerConnection(LOG, this, context, preferCachedConnections,
-        warnedAboutTServersBeingDown, RpcService.TABLET_MANAGEMENT);
+        warnedAboutTServersBeingDown, getService());
   }
 
   @Override

core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletServerThriftClient.java

@@ -48,7 +48,7 @@ public class TabletServerThriftClient extends ThriftClientTypes<Client>
   public Pair<String,Client> getThriftServerConnection(ClientContext context,
       boolean preferCachedConnections) throws TTransportException {
     return getThriftServerConnection(LOG, this, context, preferCachedConnections,
-        warnedAboutTServersBeingDown, RpcService.TSERV);
+        warnedAboutTServersBeingDown, getService());
   }
 
   @Override