You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by dl...@apache.org on 2021/03/15 19:21:59 UTC
[accumulo] branch 1451-external-compactions-feature updated: Log
status updates on the coordinator, other minor changes
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch 1451-external-compactions-feature
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/1451-external-compactions-feature by this push:
new 2b946fd Log status updates on the coordinator, other minor changes
2b946fd is described below
commit 2b946fdae9b6b7a1f686d6022e3c6f0a4e0da389
Author: Dave Marion <dl...@apache.org>
AuthorDate: Mon Mar 15 19:21:29 2021 +0000
Log status updates on the coordinator, other minor changes
---
.../java/org/apache/accumulo/core/conf/Property.java | 4 ++--
.../accumulo/coordinator/CompactionCoordinator.java | 19 +++++++++++++++----
.../java/org/apache/accumulo/compactor/Compactor.java | 13 +++++++++----
3 files changed, 26 insertions(+), 10 deletions(-)
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
index 3f77de6..bf8b610 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
@@ -1029,7 +1029,7 @@ public enum Property {
"Properties in this category affect the behavior of the accumulo compactor server."),
COMPACTOR_PORTSEARCH("compactor.port.search", "true", PropertyType.BOOLEAN,
"if the compactor.port.client is in use, search higher ports until one is available"),
- COMPACTOR_CLIENTPORT("compactor.port.client", "9100", PropertyType.PORT,
+ COMPACTOR_CLIENTPORT("compactor.port.client", "9101", PropertyType.PORT,
"The port used for handling client connections on the compactor servers"),
COMPACTOR_MINTHREADS("compactor.server.threads.minimum", "1", PropertyType.COUNT,
"The minimum number of threads to use to handle incoming requests."),
@@ -1044,7 +1044,7 @@ public enum Property {
"Properties in this category affect the behavior of the accumulo compaction coordinator server."),
COORDINATOR_PORTSEARCH("coordinator.port.search", "false", PropertyType.BOOLEAN,
"if the ports above are in use, search higher ports until one is available"),
- COORDINATOR_CLIENTPORT("coordinator.port.client", "9101", PropertyType.PORT,
+ COORDINATOR_CLIENTPORT("coordinator.port.client", "9100", PropertyType.PORT,
"The port used for handling client connections on the compactor servers"),
COORDINATOR_MINTHREADS("coordinator.server.threads.minimum", "1", PropertyType.COUNT,
"The minimum number of threads to use to handle incoming requests."),
diff --git a/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/CompactionCoordinator.java b/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/CompactionCoordinator.java
index 0a066b9..a8e2f23 100644
--- a/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/CompactionCoordinator.java
+++ b/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/CompactionCoordinator.java
@@ -80,6 +80,7 @@ public class CompactionCoordinator extends AbstractServer
private static final Logger LOG = LoggerFactory.getLogger(CompactionCoordinator.class);
private static final long TIME_BETWEEN_CHECKS = 5000;
+ private static final long TSERVER_CHECK_INTERVAL = 60000;
/* Map of external queue name -> priority -> tservers */
private static final Map<String,TreeMap<Long,LinkedHashSet<TServerInstance>>> QUEUES =
@@ -192,11 +193,13 @@ public class CompactionCoordinator extends AbstractServer
tserverSet.startListeningForTabletServerChanges();
while (true) {
+ long start = System.currentTimeMillis();
tserverSet.getCurrentServers().forEach(tsi -> {
try {
TabletClientService.Client client = null;
try {
- LOG.debug("contacting tserver " + tsi.getHostPort());
+ LOG.debug("Contacting tablet server {} to get external compaction summaries",
+ tsi.getHostPort());
client = getTabletServerConnection(tsi);
List<TCompactionQueueSummary> summaries =
client.getCompactionQueueInfo(TraceUtil.traceInfo(), getContext().rpcCreds());
@@ -213,11 +216,14 @@ public class CompactionCoordinator extends AbstractServer
ThriftUtil.returnClient(client);
}
} catch (TException e) {
- LOG.warn("Error getting compaction summaries from tablet server: {}",
+ LOG.warn("Error getting external compaction summaries from tablet server: {}",
tsi.getHostAndPort(), e);
}
});
- UtilWaitThread.sleep(60000);
+ long duration = (System.currentTimeMillis() - start);
+ if (TSERVER_CHECK_INTERVAL - duration > 0) {
+ UtilWaitThread.sleep(TSERVER_CHECK_INTERVAL - duration);
+ }
}
}
@@ -310,7 +316,8 @@ public class CompactionCoordinator extends AbstractServer
LOG.debug("No compactions found for queue {}, returning null to compactor {}", queue,
compactorAddress);
// CBUG Returning null here causes:
- // [compaction.RetryableThriftCall] ERROR: Error in Thrift function, retrying in 60000ms. Error: org.apache.thrift.TApplicationException: getCompactionJob failed: unknown result
+ // [compaction.RetryableThriftCall] ERROR: Error in Thrift function, retrying in 60000ms.
+ // Error: org.apache.thrift.TApplicationException: getCompactionJob failed: unknown result
return null;
}
@@ -353,6 +360,7 @@ public class CompactionCoordinator extends AbstractServer
*/
@Override
public void cancelCompaction(String externalCompactionId) throws TException {
+ LOG.info("Compaction cancel requested, id: {}", externalCompactionId);
RunningCompaction rc = RUNNING.get(ExternalCompactionId.of(externalCompactionId));
if (null == rc) {
throw new UnknownCompactionIdException();
@@ -406,6 +414,7 @@ public class CompactionCoordinator extends AbstractServer
@Override
public void compactionCompleted(String externalCompactionId, CompactionStats stats)
throws TException {
+ LOG.info("Compaction completed, id: {}, stats: {}", externalCompactionId, stats);
var ecid = ExternalCompactionId.of(externalCompactionId);
RunningCompaction rc = RUNNING.get(ecid);
if (null != rc) {
@@ -475,6 +484,8 @@ public class CompactionCoordinator extends AbstractServer
@Override
public void updateCompactionStatus(String externalCompactionId, CompactionState state,
String message, long timestamp) throws TException {
+ LOG.info("Compaction status update, id: {}, timestamp: {}, state: {}, message: {}",
+ externalCompactionId, timestamp, state, message);
RunningCompaction rc = RUNNING.get(ExternalCompactionId.of(externalCompactionId));
if (null != rc) {
rc.addUpdate(timestamp, message, state);
diff --git a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
index 2b8e383..503b1c5 100644
--- a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
+++ b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
@@ -55,7 +55,6 @@ import org.apache.accumulo.core.util.Halt;
import org.apache.accumulo.core.util.HostAndPort;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.core.util.threads.Threads;
-import org.apache.accumulo.fate.util.UtilWaitThread;
import org.apache.accumulo.fate.zookeeper.ZooLock;
import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason;
import org.apache.accumulo.fate.zookeeper.ZooLock.LockWatcher;
@@ -417,7 +416,7 @@ public class Compactor extends AbstractServer
throw new RuntimeException("Compaction failed", e);
} finally {
stopped.countDown();
- // TODO: Any cleanup
+ // CBUG Any cleanup
}
}
};
@@ -473,7 +472,14 @@ public class Compactor extends AbstractServer
try {
started.await(); // wait until the compactor is started
long inputEntries = totalInputEntries.sum();
- while (stopped.getCount() > 0) {
+
+ // This block of code will update the coordinator with the progress of external
+ // compaction every minute
+ //
+ // CBUG Is 1 minute too often for very large files? Since the update is stored in the
+ // RUNNING set, it's going to eat up memory. Should the time be dependent on the size
+ // of the input entries?
+ while (!stopped.await(1, TimeUnit.MINUTES)) {
List<CompactionInfo> running =
org.apache.accumulo.server.compaction.Compactor.getRunningCompactions();
if (!running.isEmpty()) {
@@ -493,7 +499,6 @@ public class Compactor extends AbstractServer
}
}
}
- UtilWaitThread.sleep(60000);
}
try {
compactionThread.join();