This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new ec74d7eaec Encapsulate log sorting task, ZooKeeper serialization
(#6601)
ec74d7eaec is described below
commit ec74d7eaecf3d7c0e0c7fb03a5bdeffd025e9a77
Author: Dave Marion <[email protected]>
AuthorDate: Fri Nov 13 10:13:12 2026 -0500
Encapsulate log sorting task, ZooKeeper serialization (#6601)
Move the serialization and deserialization of the log sorting
task to an object to make future changes easier. RecoveryManager
and LogSorter were depending on the format of the string to
be the same.
Updated the string to include the sortId and include changes in
the upgrade code to rewrite existing ZooKeeper paths to include
the sortId.
---
.../accumulo/server/log/LogSortInformation.java | 51 +++++++++++++++++++++
.../server/log/LogSortInformationTest.java | 48 ++++++++++++++++++++
.../accumulo/manager/recovery/RecoveryManager.java | 41 +++++++++--------
.../accumulo/manager/upgrade/Upgrader11to12.java | 52 ++++++++++++++++++++++
.../manager/upgrade/Upgrader11to12Test.java | 32 +++++++++++++
.../org/apache/accumulo/tserver/log/LogSorter.java | 23 +++++++---
6 files changed, 223 insertions(+), 24 deletions(-)
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/log/LogSortInformation.java
b/server/base/src/main/java/org/apache/accumulo/server/log/LogSortInformation.java
new file mode 100644
index 0000000000..eae3c6651c
--- /dev/null
+++
b/server/base/src/main/java/org/apache/accumulo/server/log/LogSortInformation.java
@@ -0,0 +1,51 @@
+/*
+ * 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.log;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.UUID;
+
+/**
+ * The information needed to sort a write-ahead log, and its representation in
the recovery work
+ * queue.
+ */
+public record LogSortInformation(UUID sortId, String source, String
destination) {
+
+ public LogSortInformation {
+ requireNonNull(sortId, "sortId is null");
+ requireNonNull(source, "source is null");
+ requireNonNull(destination, "destination is null");
+ }
+
+ /** Serialize this work item for storage in ZooKeeper. */
+ public String serialize() {
+ return sortId + "|" + source + "|" + destination;
+ }
+
+ /** Parse a serialized recovery work item. */
+ public static LogSortInformation deserialize(String serialized) {
+ requireNonNull(serialized, "serialized sort information is null");
+ String[] parts = serialized.split("\\|", -1);
+ if (parts.length != 3) {
+ throw new IllegalArgumentException("Malformed log sort work: " +
serialized);
+ }
+ return new LogSortInformation(UUID.fromString(parts[0]), parts[1],
parts[2]);
+ }
+}
diff --git
a/server/base/src/test/java/org/apache/accumulo/server/log/LogSortInformationTest.java
b/server/base/src/test/java/org/apache/accumulo/server/log/LogSortInformationTest.java
new file mode 100644
index 0000000000..6eed1d7f79
--- /dev/null
+++
b/server/base/src/test/java/org/apache/accumulo/server/log/LogSortInformationTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.log;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.UUID;
+
+import org.junit.jupiter.api.Test;
+
+class LogSortInformationTest {
+
+ @Test
+ void roundTrip() {
+ UUID sortId = UUID.randomUUID();
+ LogSortInformation work = new LogSortInformation(sortId,
"hdfs://nn/wal/another-file.wal",
+ "hdfs://nn/recovery/sorted");
+
+ assertEquals(work, LogSortInformation.deserialize(work.serialize()));
+ }
+
+ @Test
+ void malformedPayload() {
+ assertThrows(IllegalArgumentException.class,
+ () -> LogSortInformation.deserialize("source|dest"));
+ assertThrows(IllegalArgumentException.class,
+ () -> LogSortInformation.deserialize("not-a-uuid|source|dest"));
+ assertThrows(IllegalArgumentException.class,
+ () -> LogSortInformation.deserialize(UUID.randomUUID() +
"|source|dest|extra"));
+ }
+}
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
index e0cf6f2454..07fab1efdd 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
@@ -30,6 +30,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -42,6 +43,7 @@ import org.apache.accumulo.core.util.cache.Caches.CacheName;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.manager.Manager;
import org.apache.accumulo.server.fs.VolumeUtil;
+import org.apache.accumulo.server.log.LogSortInformation;
import org.apache.accumulo.server.log.SortedLogState;
import org.apache.accumulo.server.manager.recovery.HadoopLogCloser;
import org.apache.accumulo.server.manager.recovery.LogCloser;
@@ -87,10 +89,10 @@ public class RecoveryManager {
private class LogSortTask implements Runnable {
private final String source;
private final String destination;
- private final String sortId;
+ private final UUID sortId;
private final LogCloser closer;
- public LogSortTask(LogCloser closer, String source, String destination,
String sortId) {
+ public LogSortTask(LogCloser closer, String source, String destination,
UUID sortId) {
this.closer = closer;
this.source = source;
this.destination = destination;
@@ -118,7 +120,7 @@ public class RecoveryManager {
} finally {
if (!rescheduled) {
synchronized (RecoveryManager.this) {
- closeTasksQueued.remove(sortId);
+ closeTasksQueued.remove(sortId.toString());
}
}
}
@@ -126,14 +128,14 @@ public class RecoveryManager {
}
- private void initiateSort(String sortId, String source, final String
destination)
+ private void initiateSort(UUID sortId, String source, final String
destination)
throws KeeperException, InterruptedException {
- String work = source + "|" + destination;
+ String work = new LogSortInformation(sortId, source,
destination).serialize();
new DistributedWorkQueue(Constants.ZRECOVERY, manager.getConfiguration(),
manager)
- .addWork(sortId, work.getBytes(UTF_8));
+ .addWork(sortId.toString(), work.getBytes(UTF_8));
synchronized (this) {
- sortsQueued.add(sortId);
+ sortsQueued.add(sortId.toString());
}
log.info("Created zookeeper entry {} with data {}", Constants.ZRECOVERY +
"/" + sortId, work);
@@ -191,28 +193,29 @@ public class RecoveryManager {
walog = switchedWalog;
}
- String sortId = walog.getUniqueID().toString();
+ UUID sortId = walog.getUniqueID();
+ String sortIdString = sortId.toString();
String filename = walog.getPath().toString();
- String dest = RecoveryPath.getRecoveryPath(walog.getPath()).toString();
+ String dest = RecoveryPath.getRecoveryPath(new Path(filename)).toString();
boolean sortQueued;
synchronized (this) {
- sortQueued = sortsQueued.contains(sortId);
+ sortQueued = sortsQueued.contains(sortIdString);
}
if (sortQueued
- && this.manager.getContext().getZooCache().get(Constants.ZRECOVERY +
"/" + sortId)
+ && this.manager.getContext().getZooCache().get(Constants.ZRECOVERY +
"/" + sortIdString)
== null) {
synchronized (this) {
- sortsQueued.remove(sortId);
+ sortsQueued.remove(sortIdString);
}
}
if (exists(SortedLogState.getFinishedMarkerPath(dest))) {
synchronized (this) {
- closeTasksQueued.remove(sortId);
- recoveryDelay.remove(sortId);
- sortsQueued.remove(sortId);
+ closeTasksQueued.remove(sortIdString);
+ recoveryDelay.remove(sortIdString);
+ sortsQueued.remove(sortIdString);
}
return false;
// was continue;
@@ -220,11 +223,11 @@ public class RecoveryManager {
recoveryNeeded = true;
synchronized (this) {
- if (!closeTasksQueued.contains(sortId) && !sortsQueued.contains(sortId))
{
+ if (!closeTasksQueued.contains(sortIdString) &&
!sortsQueued.contains(sortIdString)) {
AccumuloConfiguration aconf = manager.getConfiguration();
LogCloser closer = Property.createInstanceFromPropertyName(aconf,
Property.MANAGER_WAL_CLOSER_IMPLEMENTATION, LogCloser.class, new
HadoopLogCloser());
- Long delay = recoveryDelay.get(sortId);
+ Long delay = recoveryDelay.get(sortIdString);
if (delay == null) {
delay = aconf.getTimeInMillis(Property.MANAGER_RECOVERY_DELAY);
} else {
@@ -236,8 +239,8 @@ public class RecoveryManager {
ScheduledFuture<?> future = executor.schedule(
new LogSortTask(closer, filename, dest, sortId), delay,
TimeUnit.MILLISECONDS);
ThreadPools.watchNonCriticalScheduledTask(future);
- closeTasksQueued.add(sortId);
- recoveryDelay.put(sortId, delay);
+ closeTasksQueued.add(sortIdString);
+ recoveryDelay.put(sortIdString, delay);
}
}
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
b/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
index cee839ee8a..b1978faf2f 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
@@ -40,6 +40,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -103,6 +104,7 @@ import org.apache.accumulo.server.conf.store.SystemPropKey;
import org.apache.accumulo.server.conf.store.TablePropKey;
import org.apache.accumulo.server.init.FileSystemInitializer;
import org.apache.accumulo.server.init.InitialConfiguration;
+import org.apache.accumulo.server.log.LogSortInformation;
import org.apache.accumulo.server.util.FindCompactionTmpFiles.DeleteStats;
import org.apache.accumulo.server.util.PropUtil;
import org.apache.hadoop.fs.FileStatus;
@@ -240,6 +242,8 @@ public class Upgrader11to12 implements Upgrader {
public void upgradeZookeeper(ServerContext context) {
LOG.info("Ensuring all worker server processes are down.");
validateEmptyZKWorkerServerPaths(context);
+ LOG.info("Updating recovery work payloads");
+ updateRecoveryWork(context);
LOG.info("Removing ZTracer node");
removeZTracersNode(context);
LOG.info("Updating file references in root tablet");
@@ -274,6 +278,54 @@ public class Upgrader11to12 implements Upgrader {
addShuttingDownTservers(context);
}
+ private void updateRecoveryWork(ServerContext context) {
+ try {
+ upgradeRecoveryWork(context.getZooSession().asReaderWriter());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while updating recovery
work payloads", e);
+ } catch (KeeperException e) {
+ throw new IllegalStateException("Could not update recovery work
payloads", e);
+ }
+ }
+
+ @VisibleForTesting
+ void upgradeRecoveryWork(ZooReaderWriter zrw) throws KeeperException,
InterruptedException {
+ for (String child : zrw.getChildren(Constants.ZRECOVERY)) {
+ if (child.equals("locks")) {
+ continue;
+ }
+
+ String childPath = Constants.ZRECOVERY + "/" + child;
+ String serialized = new String(zrw.getData(childPath), UTF_8);
+ String updated = convertRecoveryWork(child, serialized);
+ if (!serialized.equals(updated)
+ && !zrw.overwritePersistentData(childPath, updated.getBytes(UTF_8),
-1)) {
+ throw new IllegalStateException("Could not update recovery work " +
childPath);
+ }
+ }
+ }
+
+ static String convertRecoveryWork(String child, String serialized) {
+ var sortId = UUID.fromString(child);
+ String[] parts = serialized.split("\\|", -1);
+
+ // This item may already have been converted if an earlier upgrade attempt
failed partway
+ // through. Validate it and leave it unchanged so the operation is safe to
retry.
+ if (parts.length == 3) {
+ LogSortInformation work = LogSortInformation.deserialize(serialized);
+ if (!work.sortId().equals(sortId)) {
+ throw new IllegalStateException("Recovery work ID does not match child
" + child);
+ }
+ return work.serialize();
+ }
+
+ if (parts.length != 2) {
+ throw new IllegalArgumentException("Malformed legacy recovery work for "
+ child);
+ }
+ return new LogSortInformation(sortId, parts[0], parts[1]).serialize();
+ }
+
@Override
public void upgradeRoot(ServerContext context) {
LOG.info("Updating file references in metadata tablets");
diff --git
a/server/manager/src/test/java/org/apache/accumulo/manager/upgrade/Upgrader11to12Test.java
b/server/manager/src/test/java/org/apache/accumulo/manager/upgrade/Upgrader11to12Test.java
index 806ff54ab7..988a495d5a 100644
---
a/server/manager/src/test/java/org/apache/accumulo/manager/upgrade/Upgrader11to12Test.java
+++
b/server/manager/src/test/java/org/apache/accumulo/manager/upgrade/Upgrader11to12Test.java
@@ -86,6 +86,7 @@ import org.apache.accumulo.server.conf.store.SystemPropKey;
import org.apache.accumulo.server.conf.store.TablePropKey;
import org.apache.accumulo.server.conf.store.impl.ZooPropStore;
import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.accumulo.server.log.LogSortInformation;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -626,6 +627,37 @@ public class Upgrader11to12Test extends WithTestNames {
}
+ @Test
+ void testUpgradeRecoveryWork() throws InterruptedException, KeeperException {
+ Upgrader11to12 upgrader = new Upgrader11to12();
+ ZooReaderWriter zrw = createStrictMock(ZooReaderWriter.class);
+ UUID sortId = UUID.randomUUID();
+ String childPath = Constants.ZRECOVERY + "/" + sortId;
+ String source = "hdfs://localhost:8020/accumulo/wal/server/old-name.wal";
+ String destination = "hdfs://localhost:8020/accumulo/recovery/sorted";
+ String oldPayload = source + "|" + destination;
+ String newPayload = new LogSortInformation(sortId, source,
destination).serialize();
+
+
expect(zrw.getChildren(Constants.ZRECOVERY)).andReturn(List.of(sortId.toString(),
"locks"));
+ expect(zrw.getData(childPath)).andReturn(oldPayload.getBytes(UTF_8));
+ expect(zrw.overwritePersistentData(eq(childPath),
aryEq(newPayload.getBytes(UTF_8)), eq(-1)))
+ .andReturn(true);
+
+ replay(zrw);
+ upgrader.upgradeRecoveryWork(zrw);
+ verify(zrw);
+ }
+
+ @Test
+ void testConvertRecoveryWorkIsIdempotent() {
+ UUID sortId = UUID.randomUUID();
+ String payload = new LogSortInformation(sortId, "source",
"destination").serialize();
+
+ assertEquals(payload,
Upgrader11to12.convertRecoveryWork(sortId.toString(), payload));
+ assertThrows(IllegalStateException.class,
+ () -> Upgrader11to12.convertRecoveryWork(UUID.randomUUID().toString(),
payload));
+ }
+
@Test
public void testAddingTableMappingToZooKeeper() throws InterruptedException,
KeeperException {
Upgrader11to12 upgrader = new Upgrader11to12();
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java
index 93a23b52e4..25956fe296 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java
@@ -56,6 +56,7 @@ import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.server.AbstractServer;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.accumulo.server.log.LogSortInformation;
import org.apache.accumulo.server.log.SortedLogState;
import org.apache.accumulo.server.zookeeper.DistributedWorkQueue;
import org.apache.accumulo.server.zookeeper.DistributedWorkQueue.Processor;
@@ -97,11 +98,23 @@ public class LogSorter implements MetricsProducer {
@Override
public void process(String child, byte[] data) {
- String work = new String(data, UTF_8);
- String[] parts = work.split("\\|");
- String src = parts[0];
- String dest = parts[1];
- String sortId = new Path(src).getName();
+ LogSortInformation work;
+ try {
+ work = LogSortInformation.deserialize(new String(data, UTF_8));
+ } catch (IllegalArgumentException e) {
+ log.error("Discarding malformed log sort work item {}", child, e);
+ return;
+ }
+
+ if (!child.equals(work.sortId().toString())) {
+ log.error("Discarding log sort work item {} with mismatched payload
sortId {}", child,
+ work.sortId());
+ return;
+ }
+
+ String src = work.source();
+ String dest = work.destination();
+ String sortId = work.sortId().toString();
log.debug("Sorting {} to {} using sortId {}", src, dest, sortId);
synchronized (currentWork) {
