From f0c8c17ccb386c221a8c226cae2242e68178dae2 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Fri, 21 Aug 2026 12:13:52 +0000 Subject: [PATCH 1/2] Added ability to wake Compactors that might be in a long wait When there is no compaction work Compactors will wait progressively longer (between COMPACTOR_MIN_JOB_WAIT_TIME and COMPACTOR_MAX_JOB_WAIT_TIME) before checking in with the Coordinator for work. The wait time is used to reduce the number of RPC calls to the Coordinator. However, with large max wait times Compactors may sit idle when there is work to do. This change allows the user to configure the Coordinator to wake waiting Compactors when there is work in the queue. Closes #4664 --- .../apache/accumulo/core/conf/Property.java | 7 + .../compaction/ExternalCompactionUtil.java | 19 + .../core/util/threads/ThreadPoolNames.java | 1 + .../compaction/thrift/CompactorService.java | 619 ++++++++++++++++++ .../main/thrift/compaction-coordinator.thrift | 5 + .../coordinator/CompactionCoordinator.java | 34 +- .../accumulo/coordinator/QueueSummaries.java | 19 +- .../apache/accumulo/compactor/Compactor.java | 42 +- .../accumulo/compactor/CompactorTest.java | 43 ++ .../compaction/ExternalCompactionWaitIT.java | 179 +++++ 10 files changed, 957 insertions(+), 11 deletions(-) create mode 100644 test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionWaitIT.java 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 6a4caafbda7..5df6b1c1736 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 @@ -1698,6 +1698,13 @@ public enum Property { COMPACTION_COORDINATOR_TSERVER_COMPACTION_CHECK_INTERVAL( "compaction.coordinator.tserver.check.interval", "1m", PropertyType.TIMEDURATION, "The interval at which to check the tservers for external compactions.", "2.1.0"), + @Experimental + COMPACTION_COORDINATOR_COMPACTOR_WAKEUP_THREADS("compaction.coordinator.compactor.wakeup.threads", + "0", PropertyType.COUNT, + "The number of threads the Coordinator should use to wake Compactors that are in a wait state. A value of zero" + + " disables Compactor wake up. Enabling this feature will cause Compactors in a long wait state (see" + + " COMPACTOR_MAX_JOB_WAIT_TIME) to check in with the Coordinator for work.", + "2.1.7"), // deprecated properties grouped at the end to reference property that replaces them @Deprecated(since = "1.6.0") @ReplacedBy(property = INSTANCE_VOLUMES) diff --git a/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java b/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java index c6be864ce8a..28671da0e57 100644 --- a/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java +++ b/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java @@ -19,6 +19,7 @@ package org.apache.accumulo.core.util.compaction; import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.accumulo.core.util.threads.ThreadPoolNames.COMPACTION_COORDINATOR_COMPACTOR_WAKE_POOL; import static org.apache.accumulo.core.util.threads.ThreadPoolNames.COMPACTOR_RUNNING_COMPACTIONS_POOL; import static org.apache.accumulo.core.util.threads.ThreadPoolNames.COMPACTOR_RUNNING_COMPACTION_IDS_POOL; @@ -326,4 +327,22 @@ public static void cancelCompaction(ClientContext context, HostAndPort compactor ThriftUtil.returnClient(client, context); } } + + public static void wakeCompactors(ClientContext context, List compactors, + int threads) { + final ExecutorService executor = ThreadPools.getServerThreadPools() + .getPoolBuilder(COMPACTION_COORDINATOR_COMPACTOR_WAKE_POOL).numCoreThreads(threads).build(); + compactors.forEach(c -> { + CompactorService.Client client = null; + try { + client = ThriftUtil.getClient(ThriftClientTypes.COMPACTOR, c, context); + client.wake(TraceUtil.traceInfo(), context.rpcCreds()); + } catch (TException e) { + LOG.debug("Failed to wake compactor {}", c); + } finally { + ThriftUtil.returnClient(client, context); + } + }); + executor.shutdown(); + } } diff --git a/core/src/main/java/org/apache/accumulo/core/util/threads/ThreadPoolNames.java b/core/src/main/java/org/apache/accumulo/core/util/threads/ThreadPoolNames.java index 861a6ee89e6..7778262be8a 100644 --- a/core/src/main/java/org/apache/accumulo/core/util/threads/ThreadPoolNames.java +++ b/core/src/main/java/org/apache/accumulo/core/util/threads/ThreadPoolNames.java @@ -28,6 +28,7 @@ public enum ThreadPoolNames { BULK_IMPORT_CLIENT_BULK_THREADS_POOL("accumulo.pool.bulk.import.client.bulk.threads"), BULK_IMPORT_DIR_MOVE_POOL("accumulo.pool.bulk.dir.move"), COMPACTION_COORDINATOR_SUMMARY_POOL("accumulo.pool.compaction.summary.gatherer"), + COMPACTION_COORDINATOR_COMPACTOR_WAKE_POOL("accumulo.pool.compaction.compactor.wake"), COMPACTION_SERVICE_COMPACTION_PLANNER_POOL("accumulo.pool.compaction.service.compaction.planner"), COMPACTOR_RUNNING_COMPACTIONS_POOL("accumulo.pool.compactor.running.compactions"), COMPACTOR_RUNNING_COMPACTION_IDS_POOL("accumulo.pool.compactor.running.compaction.ids"), diff --git a/core/src/main/thrift-gen-java/org/apache/accumulo/core/compaction/thrift/CompactorService.java b/core/src/main/thrift-gen-java/org/apache/accumulo/core/compaction/thrift/CompactorService.java index 57cbc70eff6..de3a7210344 100644 --- a/core/src/main/thrift-gen-java/org/apache/accumulo/core/compaction/thrift/CompactorService.java +++ b/core/src/main/thrift-gen-java/org/apache/accumulo/core/compaction/thrift/CompactorService.java @@ -35,6 +35,8 @@ public interface Iface { public java.util.List getActiveCompactions(org.apache.accumulo.core.trace.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials) throws org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException, org.apache.thrift.TException; + public void wake(org.apache.accumulo.core.trace.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials) throws org.apache.thrift.TException; + public void cancel(org.apache.accumulo.core.trace.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.lang.String externalCompactionId) throws org.apache.thrift.TException; } @@ -47,6 +49,8 @@ public interface AsyncIface { public void getActiveCompactions(org.apache.accumulo.core.trace.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, org.apache.thrift.async.AsyncMethodCallback> resultHandler) throws org.apache.thrift.TException; + public void wake(org.apache.accumulo.core.trace.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException; + public void cancel(org.apache.accumulo.core.trace.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.lang.String externalCompactionId, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException; } @@ -157,6 +161,20 @@ public java.util.List resultHandler) throws org.apache.thrift.TException { + checkReady(); + wake_call method_call = new wake_call(tinfo, credentials, resultHandler, this, ___protocolFactory, ___transport); + this.___currentMethod = method_call; + ___manager.call(method_call); + } + + public static class wake_call extends org.apache.thrift.async.TAsyncMethodCall { + private org.apache.accumulo.core.trace.thrift.TInfo tinfo; + private org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials; + public wake_call(org.apache.accumulo.core.trace.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, org.apache.thrift.async.AsyncMethodCallback resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException { + super(client, protocolFactory, transport, resultHandler, true); + this.tinfo = tinfo; + this.credentials = credentials; + } + + @Override + public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException { + prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("wake", org.apache.thrift.protocol.TMessageType.ONEWAY, 0)); + wake_args args = new wake_args(); + args.setTinfo(tinfo); + args.setCredentials(credentials); + args.write(prot); + prot.writeMessageEnd(); + } + + @Override + public Void getResult() throws org.apache.thrift.TException { + if (getState() != org.apache.thrift.async.TAsyncMethodCall.State.RESPONSE_READ) { + throw new java.lang.IllegalStateException("Method call not finished!"); + } + org.apache.thrift.transport.TMemoryInputTransport memoryTransport = new org.apache.thrift.transport.TMemoryInputTransport(getFrameBuffer().array()); + org.apache.thrift.protocol.TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport); + return null; + } + } + @Override public void cancel(org.apache.accumulo.core.trace.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.lang.String externalCompactionId, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException { checkReady(); @@ -371,6 +427,7 @@ protected Processor(I iface, java.util.Map extends org.apache.thrift.ProcessFunction { + public wake() { + super("wake"); + } + + @Override + public wake_args getEmptyArgsInstance() { + return new wake_args(); + } + + @Override + protected boolean isOneway() { + return true; + } + + @Override + protected boolean rethrowUnhandledExceptions() { + return false; + } + + @Override + public org.apache.thrift.TBase getResult(I iface, wake_args args) throws org.apache.thrift.TException { + iface.wake(args.tinfo, args.credentials); + return null; + } + } + public static class cancel extends org.apache.thrift.ProcessFunction { public cancel() { super("cancel"); @@ -515,6 +599,7 @@ protected AsyncProcessor(I iface, java.util.Map extends org.apache.thrift.AsyncProcessFunction { + public wake() { + super("wake"); + } + + @Override + public wake_args getEmptyArgsInstance() { + return new wake_args(); + } + + @Override + public org.apache.thrift.async.AsyncMethodCallback getResultHandler(final org.apache.thrift.server.AbstractNonblockingServer.AsyncFrameBuffer fb, final int seqid) { + final org.apache.thrift.AsyncProcessFunction fcall = this; + return new org.apache.thrift.async.AsyncMethodCallback() { + @Override + public void onComplete(Void o) { + } + @Override + public void onError(java.lang.Exception e) { + if (e instanceof org.apache.thrift.transport.TTransportException) { + _LOGGER.error("TTransportException inside handler", e); + fb.close(); + } else { + _LOGGER.error("Exception inside oneway handler", e); + } + } + }; + } + + @Override + protected boolean isOneway() { + return true; + } + + @Override + public void start(I iface, wake_args args, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException { + iface.wake(args.tinfo, args.credentials,resultHandler); + } + } + public static class cancel extends org.apache.thrift.AsyncProcessFunction { public cancel() { super("cancel"); @@ -3797,6 +3922,500 @@ private static S scheme(org.apache. } } + @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) + public static class wake_args implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { + private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("wake_args"); + + private static final org.apache.thrift.protocol.TField TINFO_FIELD_DESC = new org.apache.thrift.protocol.TField("tinfo", org.apache.thrift.protocol.TType.STRUCT, (short)1); + private static final org.apache.thrift.protocol.TField CREDENTIALS_FIELD_DESC = new org.apache.thrift.protocol.TField("credentials", org.apache.thrift.protocol.TType.STRUCT, (short)2); + + private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new wake_argsStandardSchemeFactory(); + private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new wake_argsTupleSchemeFactory(); + + public @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.trace.thrift.TInfo tinfo; // required + public @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials; // required + + /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */ + public enum _Fields implements org.apache.thrift.TFieldIdEnum { + TINFO((short)1, "tinfo"), + CREDENTIALS((short)2, "credentials"); + + private static final java.util.Map byName = new java.util.HashMap(); + + static { + for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) { + byName.put(field.getFieldName(), field); + } + } + + /** + * Find the _Fields constant that matches fieldId, or null if its not found. + */ + @org.apache.thrift.annotation.Nullable + public static _Fields findByThriftId(int fieldId) { + switch(fieldId) { + case 1: // TINFO + return TINFO; + case 2: // CREDENTIALS + return CREDENTIALS; + default: + return null; + } + } + + /** + * Find the _Fields constant that matches fieldId, throwing an exception + * if it is not found. + */ + public static _Fields findByThriftIdOrThrow(int fieldId) { + _Fields fields = findByThriftId(fieldId); + if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!"); + return fields; + } + + /** + * Find the _Fields constant that matches name, or null if its not found. + */ + @org.apache.thrift.annotation.Nullable + public static _Fields findByName(java.lang.String name) { + return byName.get(name); + } + + private final short _thriftId; + private final java.lang.String _fieldName; + + _Fields(short thriftId, java.lang.String fieldName) { + _thriftId = thriftId; + _fieldName = fieldName; + } + + @Override + public short getThriftFieldId() { + return _thriftId; + } + + @Override + public java.lang.String getFieldName() { + return _fieldName; + } + } + + // isset id assignments + public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap; + static { + java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new java.util.EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class); + tmpMap.put(_Fields.TINFO, new org.apache.thrift.meta_data.FieldMetaData("tinfo", org.apache.thrift.TFieldRequirementType.DEFAULT, + new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, org.apache.accumulo.core.trace.thrift.TInfo.class))); + tmpMap.put(_Fields.CREDENTIALS, new org.apache.thrift.meta_data.FieldMetaData("credentials", org.apache.thrift.TFieldRequirementType.DEFAULT, + new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, org.apache.accumulo.core.securityImpl.thrift.TCredentials.class))); + metaDataMap = java.util.Collections.unmodifiableMap(tmpMap); + org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(wake_args.class, metaDataMap); + } + + public wake_args() { + } + + public wake_args( + org.apache.accumulo.core.trace.thrift.TInfo tinfo, + org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials) + { + this(); + this.tinfo = tinfo; + this.credentials = credentials; + } + + /** + * Performs a deep copy on other. + */ + public wake_args(wake_args other) { + if (other.isSetTinfo()) { + this.tinfo = new org.apache.accumulo.core.trace.thrift.TInfo(other.tinfo); + } + if (other.isSetCredentials()) { + this.credentials = new org.apache.accumulo.core.securityImpl.thrift.TCredentials(other.credentials); + } + } + + @Override + public wake_args deepCopy() { + return new wake_args(this); + } + + @Override + public void clear() { + this.tinfo = null; + this.credentials = null; + } + + @org.apache.thrift.annotation.Nullable + public org.apache.accumulo.core.trace.thrift.TInfo getTinfo() { + return this.tinfo; + } + + public wake_args setTinfo(@org.apache.thrift.annotation.Nullable org.apache.accumulo.core.trace.thrift.TInfo tinfo) { + this.tinfo = tinfo; + return this; + } + + public void unsetTinfo() { + this.tinfo = null; + } + + /** Returns true if field tinfo is set (has been assigned a value) and false otherwise */ + public boolean isSetTinfo() { + return this.tinfo != null; + } + + public void setTinfoIsSet(boolean value) { + if (!value) { + this.tinfo = null; + } + } + + @org.apache.thrift.annotation.Nullable + public org.apache.accumulo.core.securityImpl.thrift.TCredentials getCredentials() { + return this.credentials; + } + + public wake_args setCredentials(@org.apache.thrift.annotation.Nullable org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials) { + this.credentials = credentials; + return this; + } + + public void unsetCredentials() { + this.credentials = null; + } + + /** Returns true if field credentials is set (has been assigned a value) and false otherwise */ + public boolean isSetCredentials() { + return this.credentials != null; + } + + public void setCredentialsIsSet(boolean value) { + if (!value) { + this.credentials = null; + } + } + + @Override + public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) { + switch (field) { + case TINFO: + if (value == null) { + unsetTinfo(); + } else { + setTinfo((org.apache.accumulo.core.trace.thrift.TInfo)value); + } + break; + + case CREDENTIALS: + if (value == null) { + unsetCredentials(); + } else { + setCredentials((org.apache.accumulo.core.securityImpl.thrift.TCredentials)value); + } + break; + + } + } + + @org.apache.thrift.annotation.Nullable + @Override + public java.lang.Object getFieldValue(_Fields field) { + switch (field) { + case TINFO: + return getTinfo(); + + case CREDENTIALS: + return getCredentials(); + + } + throw new java.lang.IllegalStateException(); + } + + /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */ + @Override + public boolean isSet(_Fields field) { + if (field == null) { + throw new java.lang.IllegalArgumentException(); + } + + switch (field) { + case TINFO: + return isSetTinfo(); + case CREDENTIALS: + return isSetCredentials(); + } + throw new java.lang.IllegalStateException(); + } + + @Override + public boolean equals(java.lang.Object that) { + if (that instanceof wake_args) + return this.equals((wake_args)that); + return false; + } + + public boolean equals(wake_args that) { + if (that == null) + return false; + if (this == that) + return true; + + boolean this_present_tinfo = true && this.isSetTinfo(); + boolean that_present_tinfo = true && that.isSetTinfo(); + if (this_present_tinfo || that_present_tinfo) { + if (!(this_present_tinfo && that_present_tinfo)) + return false; + if (!this.tinfo.equals(that.tinfo)) + return false; + } + + boolean this_present_credentials = true && this.isSetCredentials(); + boolean that_present_credentials = true && that.isSetCredentials(); + if (this_present_credentials || that_present_credentials) { + if (!(this_present_credentials && that_present_credentials)) + return false; + if (!this.credentials.equals(that.credentials)) + return false; + } + + return true; + } + + @Override + public int hashCode() { + int hashCode = 1; + + hashCode = hashCode * 8191 + ((isSetTinfo()) ? 131071 : 524287); + if (isSetTinfo()) + hashCode = hashCode * 8191 + tinfo.hashCode(); + + hashCode = hashCode * 8191 + ((isSetCredentials()) ? 131071 : 524287); + if (isSetCredentials()) + hashCode = hashCode * 8191 + credentials.hashCode(); + + return hashCode; + } + + @Override + public int compareTo(wake_args other) { + if (!getClass().equals(other.getClass())) { + return getClass().getName().compareTo(other.getClass().getName()); + } + + int lastComparison = 0; + + lastComparison = java.lang.Boolean.compare(isSetTinfo(), other.isSetTinfo()); + if (lastComparison != 0) { + return lastComparison; + } + if (isSetTinfo()) { + lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.tinfo, other.tinfo); + if (lastComparison != 0) { + return lastComparison; + } + } + lastComparison = java.lang.Boolean.compare(isSetCredentials(), other.isSetCredentials()); + if (lastComparison != 0) { + return lastComparison; + } + if (isSetCredentials()) { + lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.credentials, other.credentials); + if (lastComparison != 0) { + return lastComparison; + } + } + return 0; + } + + @org.apache.thrift.annotation.Nullable + @Override + public _Fields fieldForId(int fieldId) { + return _Fields.findByThriftId(fieldId); + } + + @Override + public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException { + scheme(iprot).read(iprot, this); + } + + @Override + public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException { + scheme(oprot).write(oprot, this); + } + + @Override + public java.lang.String toString() { + java.lang.StringBuilder sb = new java.lang.StringBuilder("wake_args("); + boolean first = true; + + sb.append("tinfo:"); + if (this.tinfo == null) { + sb.append("null"); + } else { + sb.append(this.tinfo); + } + first = false; + if (!first) sb.append(", "); + sb.append("credentials:"); + if (this.credentials == null) { + sb.append("null"); + } else { + sb.append(this.credentials); + } + first = false; + sb.append(")"); + return sb.toString(); + } + + public void validate() throws org.apache.thrift.TException { + // check for required fields + // check for sub-struct validity + if (tinfo != null) { + tinfo.validate(); + } + if (credentials != null) { + credentials.validate(); + } + } + + private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException { + try { + write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out))); + } catch (org.apache.thrift.TException te) { + throw new java.io.IOException(te); + } + } + + private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.ClassNotFoundException { + try { + read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in))); + } catch (org.apache.thrift.TException te) { + throw new java.io.IOException(te); + } + } + + private static class wake_argsStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory { + @Override + public wake_argsStandardScheme getScheme() { + return new wake_argsStandardScheme(); + } + } + + private static class wake_argsStandardScheme extends org.apache.thrift.scheme.StandardScheme { + + @Override + public void read(org.apache.thrift.protocol.TProtocol iprot, wake_args struct) throws org.apache.thrift.TException { + org.apache.thrift.protocol.TField schemeField; + iprot.readStructBegin(); + while (true) + { + schemeField = iprot.readFieldBegin(); + if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { + break; + } + switch (schemeField.id) { + case 1: // TINFO + if (schemeField.type == org.apache.thrift.protocol.TType.STRUCT) { + struct.tinfo = new org.apache.accumulo.core.trace.thrift.TInfo(); + struct.tinfo.read(iprot); + struct.setTinfoIsSet(true); + } else { + org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); + } + break; + case 2: // CREDENTIALS + if (schemeField.type == org.apache.thrift.protocol.TType.STRUCT) { + struct.credentials = new org.apache.accumulo.core.securityImpl.thrift.TCredentials(); + struct.credentials.read(iprot); + struct.setCredentialsIsSet(true); + } else { + org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); + } + break; + default: + org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); + } + iprot.readFieldEnd(); + } + iprot.readStructEnd(); + + // check for required fields of primitive type, which can't be checked in the validate method + struct.validate(); + } + + @Override + public void write(org.apache.thrift.protocol.TProtocol oprot, wake_args struct) throws org.apache.thrift.TException { + struct.validate(); + + oprot.writeStructBegin(STRUCT_DESC); + if (struct.tinfo != null) { + oprot.writeFieldBegin(TINFO_FIELD_DESC); + struct.tinfo.write(oprot); + oprot.writeFieldEnd(); + } + if (struct.credentials != null) { + oprot.writeFieldBegin(CREDENTIALS_FIELD_DESC); + struct.credentials.write(oprot); + oprot.writeFieldEnd(); + } + oprot.writeFieldStop(); + oprot.writeStructEnd(); + } + + } + + private static class wake_argsTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory { + @Override + public wake_argsTupleScheme getScheme() { + return new wake_argsTupleScheme(); + } + } + + private static class wake_argsTupleScheme extends org.apache.thrift.scheme.TupleScheme { + + @Override + public void write(org.apache.thrift.protocol.TProtocol prot, wake_args struct) throws org.apache.thrift.TException { + org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot; + java.util.BitSet optionals = new java.util.BitSet(); + if (struct.isSetTinfo()) { + optionals.set(0); + } + if (struct.isSetCredentials()) { + optionals.set(1); + } + oprot.writeBitSet(optionals, 2); + if (struct.isSetTinfo()) { + struct.tinfo.write(oprot); + } + if (struct.isSetCredentials()) { + struct.credentials.write(oprot); + } + } + + @Override + public void read(org.apache.thrift.protocol.TProtocol prot, wake_args struct) throws org.apache.thrift.TException { + org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot; + java.util.BitSet incoming = iprot.readBitSet(2); + if (incoming.get(0)) { + struct.tinfo = new org.apache.accumulo.core.trace.thrift.TInfo(); + struct.tinfo.read(iprot); + struct.setTinfoIsSet(true); + } + if (incoming.get(1)) { + struct.credentials = new org.apache.accumulo.core.securityImpl.thrift.TCredentials(); + struct.credentials.read(iprot); + struct.setCredentialsIsSet(true); + } + } + } + + private static S scheme(org.apache.thrift.protocol.TProtocol proto) { + return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme(); + } + } + @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) public static class cancel_args implements org.apache.thrift.TBase, java.io.Serializable, Cloneable, Comparable { private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("cancel_args"); diff --git a/core/src/main/thrift/compaction-coordinator.thrift b/core/src/main/thrift/compaction-coordinator.thrift index ae6a7cce6b9..01c4d443b19 100644 --- a/core/src/main/thrift/compaction-coordinator.thrift +++ b/core/src/main/thrift/compaction-coordinator.thrift @@ -161,6 +161,11 @@ service CompactorService { ) throws ( 1:client.ThriftSecurityException sec ) + + oneway void wake( + 1:trace.TInfo tinfo + 2:security.TCredentials credentials + ) void cancel( 1:trace.TInfo tinfo 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 a389ee6b0e6..8f05bba0963 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 @@ -19,6 +19,7 @@ package org.apache.accumulo.coordinator; import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.accumulo.core.conf.Property.COMPACTION_COORDINATOR_COMPACTOR_WAKEUP_THREADS; import static org.apache.accumulo.core.conf.Property.COMPACTION_COORDINATOR_SUMMARIES_MAXTHREADS; import static org.apache.accumulo.core.util.UtilWaitThread.sleepUninterruptibly; import static org.apache.accumulo.core.util.threads.ThreadPoolNames.COMPACTION_COORDINATOR_SUMMARY_POOL; @@ -28,6 +29,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -369,13 +371,33 @@ public void run() { LOG.debug("Time spent checking compaction summaries: {}ms", (now - start)); Map> idleCompactors = getIdleCompactors(); - TIME_COMPACTOR_LAST_CHECKED.forEach((queue, lastCheckTime) -> { - if ((now - lastCheckTime) > getMissingCompactorWarningTime() - && QUEUE_SUMMARIES.isCompactionsQueued(queue) && idleCompactors.containsKey(queue)) { - LOG.warn("No compactors have checked in with coordinator for queue {} in {}ms", queue, - getMissingCompactorWarningTime()); + long compactorWarnTime = getMissingCompactorWarningTime(); + + for (Entry e : TIME_COMPACTOR_LAST_CHECKED.entrySet()) { + String queueName = e.getKey(); + Long lastCheckTime = e.getValue(); + long timeSinceLastCheck = now - lastCheckTime; + long compactionsQueued = QUEUE_SUMMARIES.numCompactionsQueued(queueName); + + if (compactionsQueued > 0) { + // If there are idle compactors, wake them + // If no idle compactors and beyond the warn time, then warn + List idle = idleCompactors.get(queueName); + int wakeThreads = + getConfiguration().getCount(COMPACTION_COORDINATOR_COMPACTOR_WAKEUP_THREADS); + if (idle != null && wakeThreads > 0) { + LOG.info("Attempting to wake {} compactors for queue {}", compactionsQueued, + queueName); + ExternalCompactionUtil.wakeCompactors(getContext(), + idle.subList(0, (int) compactionsQueued), wakeThreads); + } else if (timeSinceLastCheck > compactorWarnTime) { + LOG.warn( + "No compactors have checked in with coordinator for queue {} in {}ms. Either all compactors" + + " for this queue are busy or there are no compactors for this queue.", + queueName, getMissingCompactorWarningTime()); + } } - }); + } long checkInterval = getTServerCheckInterval(); long duration = (System.currentTimeMillis() - start); diff --git a/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/QueueSummaries.java b/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/QueueSummaries.java index 1d89cd03218..85482354487 100644 --- a/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/QueueSummaries.java +++ b/server/compaction-coordinator/src/main/java/org/apache/accumulo/coordinator/QueueSummaries.java @@ -100,12 +100,23 @@ public String toString() { } } - synchronized boolean isCompactionsQueued(String queue) { - var q = QUEUES.get(queue); + synchronized Set getQueueNames() { + return QUEUES.keySet(); + } + + /* + * returns the number of tservers that have compactions for all priorities for this queue + */ + synchronized long numCompactionsQueued(String queue) { + long compactionCount = 0; + TreeMap> q = QUEUES.get(queue); if (q == null) { - return false; + return compactionCount; + } + for (TreeSet servers : q.values()) { + compactionCount += servers.size(); } - return !q.isEmpty(); + return compactionCount; } synchronized PrioTserver getNextTserver(String queue) { 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 bc6d7e2bb78..79b260f1e59 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 @@ -97,6 +97,7 @@ import org.apache.accumulo.core.util.HostAndPort; import org.apache.accumulo.core.util.ServerServices; import org.apache.accumulo.core.util.ServerServices.Service; +import org.apache.accumulo.core.util.Timer; import org.apache.accumulo.core.util.UtilWaitThread; import org.apache.accumulo.core.util.compaction.ExternalCompactionUtil; import org.apache.accumulo.core.util.threads.ThreadPools; @@ -224,6 +225,7 @@ public String getQueueName() { private final AtomicLong cancelled = new AtomicLong(0); private final AtomicLong failed = new AtomicLong(0); private final AtomicLong terminated = new AtomicLong(0); + private final AtomicBoolean stopWaiting = new AtomicBoolean(false); protected Compactor(CompactorServerOpts opts, String[] args) { super("compactor", opts, args); @@ -798,6 +800,44 @@ private void performFailureProcessing(ConsecutiveErrorHistory errorHistory) } } + // visible for tests + protected void waitForNextCompactionCheck(int compactorCount) throws InterruptedException { + long waitMillis = getWaitTimeBetweenCompactionChecks(compactorCount); + LOG.info("Waiting {}ms before checking for next compaction job", waitMillis); + Duration waitTime = Duration.ofMillis(waitMillis); + Timer timer = Timer.startNew(); + while (!timer.hasElapsed(waitTime)) { + if (stopWaiting.compareAndSet(true, false)) { + LOG.info("Wait aborted by coordinator"); + break; + } + UtilWaitThread.sleep(250); + } + } + + // visible for tests + protected boolean shouldStopWaiting() { + return stopWaiting.get(); + } + + // visible for tests + protected boolean wakeInternal() { + LOG.debug("Wake called"); + return stopWaiting.compareAndSet(false, true); + } + + @Override + public void wake(TInfo tinfo, TCredentials credentials) { + // Don't throw exceptions, this is a oneway method + try { + if (getContext().getSecurityOperation().canPerformSystemActions(credentials)) { + wakeInternal(); + } + } catch (ThriftSecurityException e) { + LOG.error("Wake called but exception thrown", e); + } + } + @Override public void run() { @@ -855,7 +895,7 @@ public void run() { job = next.getJob(); if (!job.isSetExternalCompactionId()) { LOG.trace("No external compactions in queue {}", this.queueName); - UtilWaitThread.sleep(getWaitTimeBetweenCompactionChecks(next.getCompactorCount())); + waitForNextCompactionCheck(next.getCompactorCount()); continue; } if (!job.getExternalCompactionId().equals(currentCompactionId.get().toString())) { diff --git a/server/compactor/src/test/java/org/apache/accumulo/compactor/CompactorTest.java b/server/compactor/src/test/java/org/apache/accumulo/compactor/CompactorTest.java index c3f92f8fb5f..335a8d28897 100644 --- a/server/compactor/src/test/java/org/apache/accumulo/compactor/CompactorTest.java +++ b/server/compactor/src/test/java/org/apache/accumulo/compactor/CompactorTest.java @@ -22,6 +22,7 @@ import static org.easymock.EasyMock.expect; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import java.net.UnknownHostException; @@ -32,7 +33,12 @@ import java.util.TimerTask; import java.util.UUID; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.LongAdder; @@ -545,4 +551,41 @@ public void testCompactionWaitProperty() { PowerMock.verifyAll(); } + @Test + public void testCompactorWaitAndWake() throws InterruptedException, ExecutionException { + PowerMock.resetAll(); + PowerMock.suppress(PowerMock.methods(Halt.class, "halt")); + PowerMock.suppress(PowerMock.constructor(AbstractServer.class)); + + var conf = new ConfigurationCopy(DefaultConfiguration.getInstance()); + conf.set(Property.COMPACTOR_MIN_JOB_WAIT_TIME, "3m"); + conf.set(Property.COMPACTOR_MAX_JOB_WAIT_TIME, "5m"); + + ServerContext context = PowerMock.createNiceMock(ServerContext.class); + expect(context.getConfiguration()).andReturn(conf).anyTimes(); + + Compactor.CompactorServerOpts compactorServerOpts = + PowerMock.createNiceMock(Compactor.CompactorServerOpts.class); + expect(compactorServerOpts.getQueueName()).andReturn("default"); + + PowerMock.replayAll(); + + ScheduledExecutorService executors = Executors.newSingleThreadScheduledExecutor(); + try (var c = new SuccessfulCompactor(null, null, null, context, null, compactorServerOpts)) { + ScheduledFuture f = + executors.schedule(() -> assertTrue(c.wakeInternal()), 1, TimeUnit.SECONDS); + assertFalse(c.shouldStopWaiting()); + c.waitForNextCompactionCheck(100); + assertNull(f.get()); + assertFalse(c.shouldStopWaiting()); + f = executors.schedule(() -> assertTrue(c.wakeInternal()), 1, TimeUnit.SECONDS); + c.waitForNextCompactionCheck(100); + assertNull(f.get()); + assertFalse(c.shouldStopWaiting()); + } finally { + executors.shutdownNow(); + } + PowerMock.verifyAll(); + } + } diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionWaitIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionWaitIT.java new file mode 100644 index 00000000000..491e1de877d --- /dev/null +++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionWaitIT.java @@ -0,0 +1,179 @@ +/* + * 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.test.compaction; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.apache.accumulo.core.util.UtilWaitThread.sleepUninterruptibly; +import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE1; +import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.compact; +import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.createTable; +import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.verify; +import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.writeData; + +import java.util.EnumSet; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.accumulo.compactor.Compactor; +import org.apache.accumulo.coordinator.CompactionCoordinator; +import org.apache.accumulo.core.client.Accumulo; +import org.apache.accumulo.core.client.AccumuloClient; +import org.apache.accumulo.core.client.IteratorSetting; +import org.apache.accumulo.core.conf.Property; +import org.apache.accumulo.core.iterators.IteratorUtil; +import org.apache.accumulo.core.metrics.MetricsProducer; +import org.apache.accumulo.core.util.threads.Threads; +import org.apache.accumulo.harness.AccumuloClusterHarness; +import org.apache.accumulo.minicluster.ServerType; +import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; +import org.apache.accumulo.test.functional.SlowIterator; +import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory; +import org.apache.accumulo.test.metrics.TestStatsDSink; +import org.apache.accumulo.test.util.Wait; +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Tests that external compactions wait when there is no ompaction work and can be woken by the + * coordinator when there is work. + */ +public class ExternalCompactionWaitIT extends AccumuloClusterHarness { + private static final Logger log = LoggerFactory.getLogger(ExternalCompactionWaitIT.class); + private static final int ROWS = 10_000; + public static final int CHECKER_THREAD_SLEEP_MS = 1_000; + + private static final AtomicBoolean stopCheckerThread = new AtomicBoolean(false); + private static TestStatsDSink sink; + + @BeforeAll + public static void before() throws Exception { + sink = new TestStatsDSink(); + } + + @AfterAll + public static void after() throws Exception { + if (sink != null) { + sink.close(); + } + } + + @BeforeEach + public void setup() { + stopCheckerThread.set(false); + } + + @Override + public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration coreSite) { + ExternalCompactionTestUtils.configureMiniCluster(cfg, coreSite); + cfg.setProperty(Property.COMPACTOR_MIN_JOB_WAIT_TIME, "3m"); + cfg.setProperty(Property.COMPACTION_COORDINATOR_COMPACTOR_WAKEUP_THREADS, "1"); + cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true"); + cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY, TestStatsDRegistryFactory.class.getName()); + Map sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1", + TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort())); + cfg.setSystemProperties(sysProps); + } + + @Test + public void testWaitWakeViaMetrics() throws Exception { + String table = this.getUniqueNames(1)[0]; + + final AtomicBoolean compactorIdle = new AtomicBoolean(false); + + Thread checkerThread = getMetricsCheckerThread(compactorIdle); + + try (AccumuloClient client = + Accumulo.newClient().from(getCluster().getClientProperties()).build()) { + createTable(client, table, "cs1"); + writeData(client, table, ROWS); + + cluster.getClusterControl().startCompactors(Compactor.class, 1, QUEUE1); + cluster.getClusterControl().startCoordinator(CompactionCoordinator.class); + + checkerThread.start(); + + Wait.waitFor(() -> compactorIdle.get()); + + IteratorSetting setting = new IteratorSetting(50, "Slow", SlowIterator.class); + SlowIterator.setSleepTime(setting, 1); + client.tableOperations().attachIterator(table, setting, + EnumSet.of(IteratorUtil.IteratorScope.majc)); + log.info("Compacting table"); + compact(client, table, 2, QUEUE1, true); + + Wait.waitFor(() -> !compactorIdle.get(), SECONDS.toMillis(60)); + + log.info("Done Compacting table"); + verify(client, table, 2, ROWS); + } finally { + stopCheckerThread.set(true); + checkerThread.join(); + getCluster().getClusterControl().stopAllServers(ServerType.COMPACTOR); + getCluster().getClusterControl().stopAllServers(ServerType.COMPACTION_COORDINATOR); + } + } + + /** + * Pulls metrics from the configured sink and updates the provided variables. + * + * @param totalEntriesRead this is set to the value of the entries read metric + * @param totalEntriesWritten this is set to the value of the entries written metric + */ + private static Thread getMetricsCheckerThread(AtomicBoolean compactorIdle) { + return Threads.createNonCriticalThread("metric-tailer", () -> { + log.info("Starting metric tailer"); + + sink.getLines().clear(); + + out: while (!stopCheckerThread.get()) { + List statsDMetrics = sink.getLines(); + for (String s : statsDMetrics) { + if (stopCheckerThread.get()) { + break out; + } + TestStatsDSink.Metric metric = TestStatsDSink.parseStatsDMetric(s); + // When the tablet server flushes memory to disk that can cause metrics that may throw the + // test off, so only look for metrics from the compactor. + String process = metric.getTags().getOrDefault("process.name", "none"); + if (process.equals("compactor") + && metric.getName().equals(MetricsProducer.METRICS_SERVER_IDLE)) { + int value = Integer.parseInt(metric.getValue()); + log.debug("Found metric: {} {} with value: {}", metric.getName(), metric.getTags(), + value); + switch (metric.getName()) { + case MetricsProducer.METRICS_SERVER_IDLE: + compactorIdle.set(value == 0 ? false : true); + break; + } + } + } + sleepUninterruptibly(CHECKER_THREAD_SLEEP_MS, TimeUnit.MILLISECONDS); + } + log.info("Metric tailer thread finished"); + }); + } + +} From c5877921dde1b2e7fd3d47b995359c6bad7784c0 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Fri, 21 Aug 2026 12:48:49 +0000 Subject: [PATCH 2/2] Fix javadoc --- .../accumulo/test/compaction/ExternalCompactionWaitIT.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionWaitIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionWaitIT.java index 491e1de877d..ae0b3d13ecf 100644 --- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionWaitIT.java +++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionWaitIT.java @@ -136,11 +136,8 @@ public void testWaitWakeViaMetrics() throws Exception { } } - /** + /* * Pulls metrics from the configured sink and updates the provided variables. - * - * @param totalEntriesRead this is set to the value of the entries read metric - * @param totalEntriesWritten this is set to the value of the entries written metric */ private static Thread getMetricsCheckerThread(AtomicBoolean compactorIdle) { return Threads.createNonCriticalThread("metric-tailer", () -> {