Create prototype gRPC getCompactionJob() service for external compactions

.gitattributes

@@ -18,3 +18,4 @@
 #
 
 core/src/main/thrift-gen-java/** linguist-generated=true
+rpc/grpc/src/main/protobuf-gen-java/** linguist-generated=true

core/pom.xml

@@ -61,6 +61,10 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java</artifactId>
+    </dependency>
     <dependency>
       <groupId>commons-io</groupId>
       <artifactId>commons-io</artifactId>
@@ -69,6 +73,14 @@
       <groupId>commons-logging</groupId>
       <artifactId>commons-logging</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-netty-shaded</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.micrometer</groupId>
       <artifactId>micrometer-core</artifactId>
@@ -85,6 +97,10 @@
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-access</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.accumulo</groupId>
+      <artifactId>accumulo-grpc</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-start</artifactId>
@@ -258,12 +274,14 @@
               <excludes>
                 <exclude>.*[.]impl[.].*</exclude>
                 <exclude>.*[.]thrift[.].*</exclude>
+                <exclude>.*[.]protobuf[.].*</exclude>
                 <exclude>org[.]apache[.]accumulo[.]core[.]security[.]crypto[.].*</exclude>
               </excludes>
               <allows>
-                <!--Allow API data types to reference thrift types, but do not
-		     analyze thrift types -->
+                <!--Allow API data types to reference thrift/protobuf types, but do not
+		     analyze thrift/protobuf types -->
                 <allow>org[.]apache[.]accumulo[.].*[.]thrift[.].*</allow>
+                <allow>org[.]apache[.]accumulo[.].*[.]protobuf[.].*</allow>
                 <!--Type from hadoop used in API. If adding a new type from
 		     Hadoop to the Accumulo API ensure its annotated as
 		     stable.-->
@@ -395,6 +413,13 @@
         </executions>
       </plugin>
     </plugins>
+    <extensions>
+      <extension>
+        <groupId>kr.motd.maven</groupId>
+        <artifactId>os-maven-plugin</artifactId>
+        <version>1.7.1</version>
+      </extension>
+    </extensions>
   </build>
   <profiles>
     <profile>

core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java

@@ -102,6 +102,7 @@
 import org.apache.accumulo.core.util.tables.TableZooHelper;
 import org.apache.accumulo.core.util.threads.ThreadPools;
 import org.apache.accumulo.core.util.threads.Threads;
+import org.apache.accumulo.grpc.compaction.protobuf.PCredentials;
 import org.apache.hadoop.conf.Configuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -140,7 +141,7 @@ public class ClientContext implements AccumuloClient {
   private final Supplier<SaslConnectionParams> saslSupplier;
   private final Supplier<SslConnectionParams> sslSupplier;
   private final Supplier<ScanServerSelector> scanServerSelectorSupplier;
-  private TCredentials rpcCreds;
+  private RpcCredentialsHolder rpcCreds;
   private ThriftTransportPool thriftTransportPool;
 
   private volatile boolean closed = false;
@@ -457,13 +458,22 @@ public synchronized ConditionalWriterConfig getConditionalWriterConfig() {
    * Serialize the credentials just before initiating the RPC call
    */
   public synchronized TCredentials rpcCreds() {
+    return getRpcCreds().thrift();
+  }
+
+  public synchronized PCredentials gRpcCreds() {
+    return getRpcCreds().protobuf();
+  }
+
+  private synchronized RpcCredentialsHolder getRpcCreds() {
     ensureOpen();
     if (getCredentials().getToken().isDestroyed()) {
       rpcCreds = null;
     }
 
     if (rpcCreds == null) {
-      rpcCreds = getCredentials().toThrift(getInstanceID());
+      rpcCreds = new RpcCredentialsHolder(() -> getCredentials().toThrift(getInstanceID()),
+          () -> getCredentials().toProtobuf(getInstanceID()));
     }
 
     return rpcCreds;
@@ -1093,4 +1103,23 @@ && getConfiguration().getBoolean(Property.GENERAL_MICROMETER_CACHE_METRICS_ENABL
     return caches;
   }
 
+  private static class RpcCredentialsHolder {
+
+    private final com.google.common.base.Supplier<TCredentials> tCreds;
+    private final com.google.common.base.Supplier<PCredentials> pCreds;
+
+    RpcCredentialsHolder(com.google.common.base.Supplier<TCredentials> tCreds,
+        com.google.common.base.Supplier<PCredentials> pCreds) {
+      this.tCreds = Suppliers.memoize(tCreds);
+      this.pCreds = Suppliers.memoize(pCreds);
+    }
+
+    TCredentials thrift() {
+      return tCreds.get();
+    }
+
+    PCredentials protobuf() {
+      return pCreds.get();
+    }
+  }
 }

core/src/main/java/org/apache/accumulo/core/clientImpl/Credentials.java

@@ -30,6 +30,9 @@
 import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
 import org.apache.accumulo.core.data.InstanceId;
 import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
+import org.apache.accumulo.grpc.compaction.protobuf.PCredentials;
+
+import com.google.protobuf.ByteString;
 
 /**
  * A wrapper for internal use. This class carries the instance, principal, and authentication token
@@ -99,6 +102,18 @@ public TCredentials toThrift(InstanceId instanceID) {
     return tCreds;
   }
 
+  public PCredentials toProtobuf(InstanceId instanceID) {
+    PCredentials pCreds = PCredentials.newBuilder().setPrincipal(getPrincipal())
+        .setTokenClassName(getToken().getClass().getName())
+        .setToken(ByteString.copyFrom(AuthenticationTokenSerializer.serialize(getToken())))
+        .setInstanceId(instanceID.canonical()).build();
+    if (getToken().isDestroyed()) {
+      throw new IllegalStateException("Token has been destroyed",
+          new AccumuloSecurityException(getPrincipal(), SecurityErrorCode.TOKEN_EXPIRED));
+    }
+    return pCreds;
+  }
+
   /**
    * Converts a given thrift object to our internal Credentials representation.
    *

core/src/main/java/org/apache/accumulo/core/conf/Property.java

@@ -362,6 +362,8 @@ public enum Property {
       "Properties in this category affect the behavior of the manager server.", "2.1.0"),
   MANAGER_CLIENTPORT("manager.port.client", "9999", PropertyType.PORT,
       "The port used for handling client connections on the manager.", "1.3.5"),
+  MANAGER_GRPC_CLIENTPORT("manager.port.grpc.client", "8999", PropertyType.PORT,
+      "The port used for handling gRPC client connections on the manager.", "1.3.5"),
   MANAGER_TABLET_BALANCER("manager.tablet.balancer",
       "org.apache.accumulo.core.spi.balancer.TableLoadBalancer", PropertyType.CLASSNAME,
       "The balancer class that accumulo will use to make tablet assignment and "
@@ -1432,8 +1434,8 @@ public static boolean isValidTablePropertyKey(String key) {
   // take effect; these are always system-level properties, and not namespace or table properties
   public static final EnumSet<Property> fixedProperties = EnumSet.of(
       // port options
-      GC_PORT, MANAGER_CLIENTPORT, TSERV_CLIENTPORT, SSERV_CLIENTPORT, SSERV_PORTSEARCH,
-      COMPACTOR_PORTSEARCH, TSERV_PORTSEARCH,
+      GC_PORT, MANAGER_CLIENTPORT, MANAGER_GRPC_CLIENTPORT, TSERV_CLIENTPORT, SSERV_CLIENTPORT,
+      SSERV_PORTSEARCH, COMPACTOR_PORTSEARCH, TSERV_PORTSEARCH,
 
       // max message options
       RPC_MAX_MESSAGE_SIZE,

core/src/main/java/org/apache/accumulo/core/dataImpl/KeyExtent.java

@@ -54,9 +54,12 @@
 import org.apache.accumulo.core.util.ByteBufferUtil;
 import org.apache.accumulo.core.util.Pair;
 import org.apache.accumulo.core.util.TextUtil;
+import org.apache.accumulo.grpc.compaction.protobuf.PKeyExtent;
 import org.apache.hadoop.io.BinaryComparable;
 import org.apache.hadoop.io.Text;
 
+import com.google.protobuf.ByteString;
+
 /**
  * keeps track of information needed to identify a tablet
  */
@@ -128,6 +131,32 @@ public TKeyExtent toThrift() {
         prevEndRow() == null ? null : TextUtil.getByteBuffer(prevEndRow()));
   }
 
+  /**
+   * Convert to Protobuf form.
+   */
+  public PKeyExtent toProtobuf() {
+    PKeyExtent.Builder builder = PKeyExtent.newBuilder().setTable(tableId().canonical());
+    if (endRow() != null) {
+      builder.setEndRow(ByteString.copyFrom(endRow().getBytes()));
+    }
+    if (prevEndRow() != null) {
+      builder.setPrevEndRow(ByteString.copyFrom(prevEndRow().getBytes()));
+    }
+    return builder.build();
+  }
+
+  /**
+   * Create a KeyExtent from its Protobuf form.
+   *
+   * @param pke the KeyExtent in its Protobuf object form
+   */
+  public static KeyExtent fromProtobuf(PKeyExtent pke) {
+    TableId tableId = TableId.of(pke.getTable());
+    Text endRow = !pke.hasEndRow() ? null : new Text(pke.getEndRow().toByteArray());
+    Text prevEndRow = !pke.hasPrevEndRow() ? null : new Text(pke.getPrevEndRow().toByteArray());
+    return new KeyExtent(tableId, endRow, prevEndRow);
+  }
+
   /**
    * Create a KeyExtent from a metadata previous end row entry.
    *

core/src/main/java/org/apache/accumulo/core/iteratorsImpl/system/SystemIteratorUtil.java

@@ -30,32 +30,32 @@
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.security.Authorizations;
-import org.apache.accumulo.core.tabletserver.thrift.IteratorConfig;
-import org.apache.accumulo.core.tabletserver.thrift.TIteratorSetting;
+import org.apache.accumulo.grpc.compaction.protobuf.PIteratorConfig;
+import org.apache.accumulo.grpc.compaction.protobuf.PIteratorSetting;
 
 /**
  * System utility class. Not for client use.
  */
 public class SystemIteratorUtil {
 
-  public static TIteratorSetting toTIteratorSetting(IteratorSetting is) {
-    return new TIteratorSetting(is.getPriority(), is.getName(), is.getIteratorClass(),
-        is.getOptions());
+  public static PIteratorSetting toPIteratorSetting(IteratorSetting is) {
+    return PIteratorSetting.newBuilder().setPriority(is.getPriority()).setName(is.getName())
+        .setIteratorClass(is.getIteratorClass()).putAllProperties(is.getOptions()).build();
   }
 
-  public static IteratorSetting toIteratorSetting(TIteratorSetting tis) {
-    return new IteratorSetting(tis.getPriority(), tis.getName(), tis.getIteratorClass(),
-        tis.getProperties());
+  public static IteratorSetting toIteratorSetting(PIteratorSetting pis) {
+    return new IteratorSetting(pis.getPriority(), pis.getName(), pis.getIteratorClass(),
+        pis.getPropertiesMap());
   }
 
-  public static IteratorConfig toIteratorConfig(List<IteratorSetting> iterators) {
-    ArrayList<TIteratorSetting> tisList = new ArrayList<>();
+  public static PIteratorConfig toIteratorConfig(List<IteratorSetting> iterators) {
+    ArrayList<PIteratorSetting> pisList = new ArrayList<>();
 
     for (IteratorSetting iteratorSetting : iterators) {
-      tisList.add(toTIteratorSetting(iteratorSetting));
+      pisList.add(toPIteratorSetting(iteratorSetting));
     }
 
-    return new IteratorConfig(tisList);
+    return PIteratorConfig.newBuilder().addAllIterators(pisList).build();
   }
 
   public static SortedKeyValueIterator<Key,Value> setupSystemScanIterators(