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..ae0b3d13ecf --- /dev/null +++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionWaitIT.java @@ -0,0 +1,176 @@ +/* + * 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. + */ + 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"); + }); + } + +}