Merge branch 'main' into accumulo-5014-api2

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

@@ -229,7 +229,8 @@ public ClientContext(SingletonReservation reservation, ClientInfo info,
 
     this.zooSession = memoize(() -> {
       var zk = info
-          .getZooKeeperSupplier(getClass().getSimpleName() + "(" + info.getPrincipal() + ")").get();
+          .getZooKeeperSupplier(getClass().getSimpleName() + "(" + info.getPrincipal() + ")", "")
+          .get();
       zooKeeperOpened.set(true);
       return zk;
     });
@@ -1061,7 +1062,7 @@ public synchronized ZookeeperLockChecker getTServerLockChecker() {
       // so, it can't rely on being able to continue to use the same client's ZooCache,
       // because that client could be closed, and its ZooSession also closed
       // this needs to be fixed; TODO https://github.com/apache/accumulo/issues/2301
-      var zk = info.getZooKeeperSupplier(ZookeeperLockChecker.class.getSimpleName()).get();
+      var zk = info.getZooKeeperSupplier(ZookeeperLockChecker.class.getSimpleName(), "").get();
       this.zkLockChecker = new ZookeeperLockChecker(new ZooCache(zk), getZooKeeperRoot());
     }
     return this.zkLockChecker;

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

@@ -50,7 +50,7 @@ public interface ClientInfo {
   /**
    * @return a Supplier for creating new ZooKeeper client instances based on the configuration
    */
-  Supplier<ZooSession> getZooKeeperSupplier(String clientName);
+  Supplier<ZooSession> getZooKeeperSupplier(String clientName, String rootPath);
 
   /**
    * @return Zookeeper connection information for Accumulo instance

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

@@ -29,7 +29,7 @@
 import java.nio.file.Paths;
 import java.util.Optional;
 import java.util.Properties;
-import java.util.function.Function;
+import java.util.function.BiFunction;
 import java.util.function.Supplier;
 
 import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
@@ -52,18 +52,19 @@ public class ClientInfoImpl implements ClientInfo {
   private final Supplier<AuthenticationToken> tokenSupplier;
   private final Supplier<Configuration> hadoopConf;
   private final Supplier<InstanceId> instanceId;
-  private final Function<String,ZooSession> zooSessionForName;
+  private final BiFunction<String,String,ZooSession> zooSessionForName;
 
   public ClientInfoImpl(Properties properties, Optional<AuthenticationToken> tokenOpt) {
     this.properties = requireNonNull(properties);
     // convert the optional to a supplier to delay retrieval from the properties unless needed
     this.tokenSupplier = requireNonNull(tokenOpt).map(Suppliers::ofInstance)
         .orElse(memoize(() -> ClientProperty.getAuthenticationToken(properties)));
     this.hadoopConf = memoize(Configuration::new);
-    this.zooSessionForName =
-        name -> new ZooSession(name, getZooKeepers(), getZooKeepersSessionTimeOut(), null);
+    this.zooSessionForName = (name, rootPath) -> new ZooSession(name, getZooKeepers() + rootPath,
+        getZooKeepersSessionTimeOut(), null);
     this.instanceId = memoize(() -> {
-      try (var zk = getZooKeeperSupplier(getClass().getSimpleName() + ".getInstanceName()").get()) {
+      try (var zk =
+          getZooKeeperSupplier(getClass().getSimpleName() + ".getInstanceId()", "").get()) {
         return ZooUtil.getInstanceId(zk, getInstanceName());
       }
     });
@@ -80,8 +81,8 @@ public InstanceId getInstanceId() {
   }
 
   @Override
-  public Supplier<ZooSession> getZooKeeperSupplier(String clientName) {
-    return () -> zooSessionForName.apply(clientName);
+  public Supplier<ZooSession> getZooKeeperSupplier(String clientName, String rootPath) {
+    return () -> zooSessionForName.apply(requireNonNull(clientName), requireNonNull(rootPath));
   }
 
   @Override

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

@@ -28,9 +28,11 @@
 import java.io.DataOutput;
 import java.io.DataOutputStream;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.Comparator;
@@ -55,6 +57,8 @@
 import org.apache.accumulo.core.util.Pair;
 import org.apache.accumulo.core.util.TextUtil;
 import org.apache.hadoop.io.BinaryComparable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
 import org.apache.hadoop.io.Text;
 
 /**
@@ -560,4 +564,26 @@ public String obscured() {
     return Base64.getEncoder().encodeToString(digester.digest());
   }
 
+  public String toBase64() {
+    DataOutputBuffer buffer = new DataOutputBuffer();
+    try {
+      writeTo(buffer);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+
+    return Base64.getEncoder().encodeToString(Arrays.copyOf(buffer.getData(), buffer.getLength()));
+  }
+
+  public static KeyExtent fromBase64(String encoded) {
+    byte[] data = Base64.getDecoder().decode(encoded);
+    DataInputBuffer buffer = new DataInputBuffer();
+    buffer.reset(data, data.length);
+    try {
+      return KeyExtent.readFrom(buffer);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
 }

core/src/main/java/org/apache/accumulo/core/lock/ServiceLock.java

@@ -18,7 +18,6 @@
  */
 package org.apache.accumulo.core.lock;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
@@ -745,33 +744,6 @@ public static void deleteLock(ZooReaderWriter zk, ServiceLockPath path)
 
   }
 
-  public static boolean deleteLock(ZooReaderWriter zk, ServiceLockPath path, String lockData)
-      throws InterruptedException, KeeperException {
-
-    List<String> children = validateAndSort(path, zk.getChildren(path.toString()));
-
-    if (children.isEmpty()) {
-      throw new IllegalStateException("No lock is held at " + path);
-    }
-
-    String lockNode = children.get(0);
-
-    if (!lockNode.startsWith(ZLOCK_PREFIX)) {
-      throw new IllegalStateException("Node " + lockNode + " at " + path + " is not a lock node");
-    }
-
-    byte[] data = zk.getData(path + "/" + lockNode);
-
-    if (lockData.equals(new String(data, UTF_8))) {
-      String pathToDelete = path + "/" + lockNode;
-      LOG.debug("Deleting all at path {} due to lock deletion", pathToDelete);
-      zk.recursiveDelete(pathToDelete, NodeMissingPolicy.FAIL);
-      return true;
-    }
-
-    return false;
-  }
-
   /**
    * Checks that the lock still exists in ZooKeeper. The typical mechanism for determining if a lock
    * is lost depends on a Watcher set on the lock node. There exists a case where the Watcher may

core/src/main/java/org/apache/accumulo/core/logging/TabletLogger.java

@@ -36,7 +36,6 @@
 import org.apache.accumulo.core.metadata.TServerInstance;
 import org.apache.accumulo.core.metadata.TabletFile;
 import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
-import org.apache.accumulo.core.metadata.schema.TabletMetadata;
 import org.apache.accumulo.core.spi.compaction.CompactionJob;
 import org.apache.accumulo.core.spi.compaction.CompactionKind;
 import org.apache.accumulo.core.tabletserver.log.LogEntry;
@@ -147,20 +146,18 @@ public static void selected(FateId fateId, KeyExtent extent,
         Collections2.transform(inputs, StoredTabletFile::toMinimalString));
   }
 
-  public static void compacting(TabletMetadata tabletMetadata, ExternalCompactionId cid,
+  public static void compacting(KeyExtent extent, FateId selectedFateId, ExternalCompactionId cid,
       String compactorAddress, CompactionJob job) {
     if (fileLog.isDebugEnabled()) {
       if (job.getKind() == CompactionKind.USER) {
-        var fateId = tabletMetadata.getSelectedFiles().getFateId();
         fileLog.debug(
             "Compacting {} driver:{} id:{} group:{} compactor:{} priority:{} size:{} kind:{} files:{}",
-            tabletMetadata.getExtent(), fateId, cid, job.getGroup(), compactorAddress,
-            job.getPriority(), getSize(job.getFiles()), job.getKind(),
-            asMinimalString(job.getFiles()));
+            extent, selectedFateId, cid, job.getGroup(), compactorAddress, job.getPriority(),
+            getSize(job.getFiles()), job.getKind(), asMinimalString(job.getFiles()));
       } else {
         fileLog.debug(
             "Compacting {} id:{} group:{} compactor:{} priority:{} size:{} kind:{} files:{}",
-            tabletMetadata.getExtent(), cid, job.getGroup(), compactorAddress, job.getPriority(),
+            extent, cid, job.getGroup(), compactorAddress, job.getPriority(),
             getSize(job.getFiles()), job.getKind(), asMinimalString(job.getFiles()));
       }
     }

core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java

@@ -534,9 +534,10 @@ ConditionalTabletMutator requireSame(TabletMetadata tabletMetadata, ColumnType t
     ConditionalTabletMutator requireAbsentLoaded(Set<ReferencedTabletFile> files);
 
     /**
-     * Requires the given set of files are not currently involved in any running compactions.
+     * This check will run atomically on the server side and must pass in order for the tablet to be
+     * updated.
      */
-    ConditionalTabletMutator requireNotCompacting(Set<StoredTabletFile> files);
+    ConditionalTabletMutator requireCheckSuccess(TabletMetadataCheck check);
 
     /**
      * <p>

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadataCheck.java

@@ -0,0 +1,57 @@
+/*
+ * 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.core.metadata.schema;
+
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.Set;
+
+/**
+ * This interface facilitates atomic checks of tablet metadata prior to updating tablet metadata.
+ * The way it is intended to be used is the following.
+ * <ol>
+ * <li>On the client side a TabletMetadataCheck object is created and passed to Ample</li>
+ * <li>Ample uses Gson to serialize the object, so it must not reference anything that can not
+ * serialize. Also it should not reference big things like server context or the Manager, it should
+ * only reference a small amount of data needed for the check.
+ * <li>On the tablet server side, as part of conditional mutation processing, this class is
+ * recreated and the {@link #canUpdate(TabletMetadata)} method is called and if it returns true the
+ * conditional mutation will go through.</li>
+ * </ol>
+ *
+ * <p>
+ * Implementations are expected to have a no arg constructor.
+ * </p>
+ *
+ */
+public interface TabletMetadataCheck {
+
+  Set<TabletMetadata.ColumnType> ALL_COLUMNS =
+      Collections.unmodifiableSet(EnumSet.noneOf(TabletMetadata.ColumnType.class));
+
+  boolean canUpdate(TabletMetadata tabletMetadata);
+
+  /**
+   * Determines what tablet metadata columns are read on the server side. Return the empty set to
+   * read all tablet metadata columns.
+   */
+  default Set<TabletMetadata.ColumnType> columnsToRead() {
+    return ALL_COLUMNS;
+  }
+}

core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionJobImpl.java

@@ -20,7 +20,6 @@
 
 import java.util.Collection;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
 
 import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
@@ -40,22 +39,13 @@ public class CompactionJobImpl implements CompactionJob {
   private final CompactorGroupId group;
   private final Set<CompactableFile> files;
   private final CompactionKind kind;
-  // Tracks if a job selected all the tablet's files that existed at the time the job was created.
-  private final Optional<Boolean> jobSelectedAll;
 
-  /**
-   *
-   * @param jobSelectedAll This parameters only needs to be non-empty for job objects that are used
-   *        to start compaction. After a job is running, its not used. So when a job object is
-   *        recreated for a running external compaction this parameter can be empty.
-   */
   public CompactionJobImpl(short priority, CompactorGroupId group,
-      Collection<CompactableFile> files, CompactionKind kind, Optional<Boolean> jobSelectedAll) {
+      Collection<CompactableFile> files, CompactionKind kind) {
     this.priority = priority;
     this.group = Objects.requireNonNull(group);
     this.files = Set.copyOf(files);
     this.kind = Objects.requireNonNull(kind);
-    this.jobSelectedAll = Objects.requireNonNull(jobSelectedAll);
   }
 
   @Override
@@ -92,10 +82,6 @@ public int hashCode() {
     return Objects.hash(priority, group, files, kind);
   }
 
-  public boolean selectedAll() {
-    return jobSelectedAll.orElseThrow();
-  }
-
   @Override
   public boolean equals(Object o) {
     if (o instanceof CompactionJobImpl) {

core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionPlanImpl.java

@@ -23,7 +23,6 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Optional;
 import java.util.Set;
 
 import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
@@ -80,8 +79,7 @@ public Builder addJob(short priority, CompactorGroupId group,
 
       seenFiles.addAll(filesSet);
 
-      jobs.add(new CompactionJobImpl(priority, group, filesSet, kind,
-          Optional.of(filesSet.equals(allFiles))));
+      jobs.add(new CompactionJobImpl(priority, group, filesSet, kind));
       return this;
     }
 

core/src/test/java/org/apache/accumulo/core/spi/compaction/RatioBasedCompactionPlannerTest.java

@@ -34,7 +34,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.IntStream;
 
@@ -632,7 +631,7 @@ public void testMaxTabletFilesNoCompaction() {
     // that a compaction is not planned
     all = createCFs(1_000, 2, 2, 2, 2, 2, 2, 2);
     var job = new CompactionJobImpl((short) 1, CompactorGroupId.of("ee1"), createCFs("F1", "1000"),
-        CompactionKind.SYSTEM, Optional.of(false));
+        CompactionKind.SYSTEM);
     params = createPlanningParams(all, all, Set.of(job), 3, CompactionKind.SYSTEM, conf);
     plan = planner.makePlan(params);
 

core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionPrioritizerTest.java

@@ -37,7 +37,6 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
-import java.util.Optional;
 
 import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
 import org.apache.accumulo.core.clientImpl.Namespace;
@@ -60,10 +59,9 @@ public CompactionJob createJob(CompactionKind kind, String tablet, int numFiles,
       files.add(CompactableFile
           .create(URI.create("hdfs://foonn/accumulo/tables/5/" + tablet + "/" + i + ".rf"), 4, 4));
     }
-    return new CompactionJobImpl(
-        CompactionJobPrioritizer.createPriority(Namespace.DEFAULT.id(), TableId.of("5"), kind,
-            totalFiles, numFiles, totalFiles * 2),
-        CompactorGroupId.of("test"), files, kind, Optional.of(false));
+    return new CompactionJobImpl(CompactionJobPrioritizer.createPriority(Namespace.DEFAULT.id(),
+        TableId.of("5"), kind, totalFiles, numFiles, totalFiles * 2), CompactorGroupId.of("test"),
+        files, kind);
   }
 
   @Test