Implement getCompactionJob async RPC using Jetty and REST

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

@@ -1155,7 +1155,11 @@ public enum Property {
   @Experimental
   COMPACTION_COORDINATOR_DEAD_COMPACTOR_CHECK_INTERVAL(
       "compaction.coordinator.compactor.dead.check.interval", "5m", PropertyType.TIMEDURATION,
-      "The interval at which to check for dead compactors.", "2.1.0");
+      "The interval at which to check for dead compactors.", "2.1.0"),
+  COMPACTION_COORDINATOR_MAX_JOB_REQUEST_WAIT_TIME(
+      "compaction.coordinator.wait.time.job.request.max", "2m", PropertyType.TIMEDURATION,
+      "The maximum amount of time the coordinator will wait for a requested job from the job queue.",
+      "4.0.0");
 
   private final String key;
   private final String defaultValue;

pom.xml

@@ -925,6 +925,8 @@
                 <unused>org.glassfish.jersey.ext:jersey-bean-validation:jar:*</unused>
                 <unused>org.glassfish.jersey.inject:jersey-hk2:jar:*</unused>
                 <unused>org.glassfish.jersey.test-framework.providers:jersey-test-framework-provider-grizzly2:jar:*</unused>
+                <!-- Required dependency for async servlet support -->
+                <unused>org.glassfish.jersey.containers:jersey-container-servlet:jar:*</unused>
                 <unused>org.powermock:powermock-api-easymock:jar:*</unused>
                 <!-- spotbugs annotations may or may not be used in each module -->
                 <unused>com.github.spotbugs:spotbugs-annotations:jar:*</unused>

server/base/pom.xml

@@ -35,6 +35,14 @@
       <groupId>com.beust</groupId>
       <artifactId>jcommander</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.github.ben-manes.caffeine</groupId>
       <artifactId>caffeine</artifactId>

server/base/src/main/java/org/apache/accumulo/server/rest/ThriftDeserializer.java

@@ -0,0 +1,83 @@
+/*
+ * 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.rest;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.accumulo.server.rest.ThriftSerializer.ENCODED;
+import static org.apache.accumulo.server.rest.ThriftSerializer.TYPE;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+
+import org.apache.thrift.TBase;
+import org.apache.thrift.TDeserializer;
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TJSONProtocol;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * Jackson deserializer for thrift objects that delegates the deserialization of thrift to
+ * {@link TDeserializer}. It handles previously encoded serialized objects from
+ * {@link ThriftSerializer}
+ */
+public class ThriftDeserializer<T extends TBase<?,?>> extends JsonDeserializer<T> {
+  @Override
+  public T deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+    JsonNode tree = p.readValueAsTree();
+
+    try {
+      var thriftClassName = tree.get(TYPE).asText();
+      var encoded = tree.get(ENCODED).asText();
+
+      Constructor<T> constructor = getThriftClass(thriftClassName).getDeclaredConstructor();
+      T obj = constructor.newInstance();
+      deserialize(obj, encoded);
+
+      return obj;
+    } catch (ReflectiveOperationException e) {
+      throw new IOException(e);
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private Class<T> getThriftClass(String className) throws ClassNotFoundException {
+    var clazz = Class.forName(className, false, ThriftDeserializer.class.getClassLoader());
+    // Note: This check is important to prevent potential security issues
+    // We don't want to allow arbitrary classes to be loaded
+    if (!TBase.class.isAssignableFrom(clazz)) {
+      throw new IllegalArgumentException("Class " + clazz + " is not assignable to TBase");
+    }
+    return (Class<T>) clazz;
+  }
+
+  // TODO: It doesn't seem like TDeserializer is thread safe, is there a way
+  // to prevent creating a new deserializer for every object?
+  private static <T extends TBase<?,?>> void deserialize(T obj, String json) throws IOException {
+    try {
+      final TDeserializer deserializer = new TDeserializer(new TJSONProtocol.Factory());
+      deserializer.deserialize(obj, json, UTF_8.name());
+    } catch (TException e) {
+      throw new IOException(e);
+    }
+  }
+}

server/base/src/main/java/org/apache/accumulo/server/rest/ThriftMixIn.java

@@ -0,0 +1,28 @@
+/*
+ * 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.rest;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+@JsonSerialize(using = ThriftSerializer.class)
+@JsonDeserialize(using = ThriftDeserializer.class)
+public abstract class ThriftMixIn {
+
+}

server/base/src/main/java/org/apache/accumulo/server/rest/ThriftSerializer.java

@@ -0,0 +1,60 @@
+/*
+ * 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.rest;
+
+import java.io.IOException;
+
+import org.apache.thrift.TBase;
+import org.apache.thrift.TException;
+import org.apache.thrift.TSerializer;
+import org.apache.thrift.protocol.TJSONProtocol;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+
+/**
+ * Jackson serializer for thrift objects that delegates the serialization of thrift to
+ * {@link TSerializer} and also includes the serialized type for the deserializer to use
+ */
+public class ThriftSerializer<T extends TBase<?,?>> extends JsonSerializer<T> {
+
+  static final String TYPE = "type";
+  static final String ENCODED = "encoded";
+
+  @Override
+  public void serialize(T value, JsonGenerator gen, SerializerProvider serializers)
+      throws IOException {
+    gen.writeStartObject();
+    gen.writeObjectField(TYPE, value.getClass());
+    gen.writeStringField(ENCODED, serialize(value));
+    gen.writeEndObject();
+  }
+
+  // TODO: It doesn't seem like TSerializer is thread safe, is there a way
+  // to prevent creating a new serializer for every object?
+  private static <T extends TBase<?,?>> String serialize(T obj) throws IOException {
+    try {
+      final TSerializer serializer = new TSerializer(new TJSONProtocol.Factory());
+      return serializer.toString(obj);
+    } catch (TException e) {
+      throw new IOException(e);
+    }
+  }
+}

server/base/src/main/java/org/apache/accumulo/server/rest/request/GetCompactionJobRequest.java

@@ -0,0 +1,82 @@
+/*
+ * 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.rest.request;
+
+import org.apache.accumulo.core.clientImpl.thrift.TInfo;
+import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
+
+public class GetCompactionJobRequest {
+
+  private TInfo tinfo;
+  private TCredentials credentials;
+  private String groupName;
+  private String compactorAddress;
+  private String externalCompactionId;
+
+  public GetCompactionJobRequest() {}
+
+  public GetCompactionJobRequest(TInfo tinfo, TCredentials credentials, String groupName,
+      String compactorAddress, String externalCompactionId) {
+    this.tinfo = tinfo;
+    this.credentials = credentials;
+    this.groupName = groupName;
+    this.compactorAddress = compactorAddress;
+    this.externalCompactionId = externalCompactionId;
+  }
+
+  public TInfo getTinfo() {
+    return tinfo;
+  }
+
+  public void setTinfo(TInfo tinfo) {
+    this.tinfo = tinfo;
+  }
+
+  public TCredentials getCredentials() {
+    return credentials;
+  }
+
+  public void setCredentials(TCredentials credentials) {
+    this.credentials = credentials;
+  }
+
+  public String getGroupName() {
+    return groupName;
+  }
+
+  public void setGroupName(String groupName) {
+    this.groupName = groupName;
+  }
+
+  public String getCompactorAddress() {
+    return compactorAddress;
+  }
+
+  public void setCompactorAddress(String compactorAddress) {
+    this.compactorAddress = compactorAddress;
+  }
+
+  public String getExternalCompactionId() {
+    return externalCompactionId;
+  }
+
+  public void setExternalCompactionId(String externalCompactionId) {
+    this.externalCompactionId = externalCompactionId;
+  }
+}

server/base/src/main/java/org/apache/accumulo/server/security/SystemCredentials.java

@@ -103,7 +103,7 @@ public static final class SystemToken extends PasswordToken {
      */
     public SystemToken() {}
 
-    private SystemToken(byte[] systemPassword) {
+    public SystemToken(byte[] systemPassword) {
       super(systemPassword);
     }
 

server/compactor/pom.xml

@@ -30,6 +30,10 @@
   <artifactId>accumulo-compactor</artifactId>
   <name>Apache Accumulo Compactor</name>
   <dependencies>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.auto.service</groupId>
       <artifactId>auto-service</artifactId>
@@ -71,6 +75,22 @@
       <groupId>org.apache.zookeeper</groupId>
       <artifactId>zookeeper</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-client</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-http</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-io</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-util</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>