diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/RootTable.java b/core/src/main/java/org/apache/accumulo/core/metadata/RootTable.java index 7c00cc0e815..ce3c6549088 100644 --- a/core/src/main/java/org/apache/accumulo/core/metadata/RootTable.java +++ b/core/src/main/java/org/apache/accumulo/core/metadata/RootTable.java @@ -43,6 +43,10 @@ public class RootTable { * ZK path relative to the zookeeper node where the root tablet gc candidates are stored. */ public static final String ZROOT_TABLET_GC_CANDIDATES = ZROOT_TABLET + "/gc_candidates"; + /* + * ZK path relative to the zookeeper node where the root tablet refresh entries are stored. + */ + public static final String ZROOT_TABLET_REFRESHES = ZROOT_TABLET + "/refreshes"; public static final KeyExtent EXTENT = new KeyExtent(ID, null, null); public static final KeyExtent OLD_EXTENT = diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java index 78a047dabed..9f88a69c076 100644 --- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java +++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java @@ -21,6 +21,7 @@ import java.util.Collection; import java.util.Iterator; import java.util.Map; +import java.util.Objects; import java.util.UUID; import java.util.function.Predicate; import java.util.stream.Stream; @@ -593,4 +594,46 @@ default void addBulkLoadInProgressFlag(String path, long fateTxid) { default void removeBulkLoadInProgressFlag(String path) { throw new UnsupportedOperationException(); } + + interface Refreshes { + static class RefreshEntry { + private final ExternalCompactionId ecid; + + private final KeyExtent extent; + private final TServerInstance tserver; + + public RefreshEntry(ExternalCompactionId ecid, KeyExtent extent, TServerInstance tserver) { + this.ecid = Objects.requireNonNull(ecid); + this.extent = Objects.requireNonNull(extent); + this.tserver = Objects.requireNonNull(tserver); + } + + public ExternalCompactionId getEcid() { + return ecid; + } + + public KeyExtent getExtent() { + return extent; + } + + public TServerInstance getTserver() { + return tserver; + } + } + + void add(Collection entries); + + void delete(Collection entries); + + Stream stream(); + } + + /** + * Refresh entries in the metadata table are used to track hosted tablets that need to have their + * metadata refreshed after a compaction. These entries ensure the refresh happens even in the + * case of process death. + */ + default Refreshes refreshes(DataLevel dataLevel) { + throw new UnsupportedOperationException(); + } } diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/MetadataSchema.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/MetadataSchema.java index d79a1ac9569..bb89d408784 100644 --- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/MetadataSchema.java +++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/MetadataSchema.java @@ -491,4 +491,17 @@ public static String getRowPrefix() { return section.getRowPrefix(); } } + + public static class RefreshSection { + private static final Section section = + new Section(RESERVED_PREFIX + "refresh", true, RESERVED_PREFIX + "refresi", false); + + public static Range getRange() { + return section.getRange(); + } + + public static String getRowPrefix() { + return section.getRowPrefix(); + } + } } diff --git a/core/src/main/thrift-gen-java/org/apache/accumulo/core/tabletserver/thrift/TTabletRefresh.java b/core/src/main/thrift-gen-java/org/apache/accumulo/core/tabletserver/thrift/TTabletRefresh.java new file mode 100644 index 00000000000..eb59bdc17fd --- /dev/null +++ b/core/src/main/thrift-gen-java/org/apache/accumulo/core/tabletserver/thrift/TTabletRefresh.java @@ -0,0 +1,566 @@ +/* + * 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. + */ +/** + * Autogenerated by Thrift Compiler (0.17.0) + * + * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING + * @generated + */ +package org.apache.accumulo.core.tabletserver.thrift; + +@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"}) +public class TTabletRefresh 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("TTabletRefresh"); + + private static final org.apache.thrift.protocol.TField EXTENT_FIELD_DESC = new org.apache.thrift.protocol.TField("extent", org.apache.thrift.protocol.TType.STRUCT, (short)1); + private static final org.apache.thrift.protocol.TField SCAN_ENTRIES_FIELD_DESC = new org.apache.thrift.protocol.TField("scanEntries", org.apache.thrift.protocol.TType.LIST, (short)2); + + private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new TTabletRefreshStandardSchemeFactory(); + private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new TTabletRefreshTupleSchemeFactory(); + + public @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TKeyExtent extent; // required + public @org.apache.thrift.annotation.Nullable java.util.List scanEntries; // 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 { + EXTENT((short)1, "extent"), + SCAN_ENTRIES((short)2, "scanEntries"); + + 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: // EXTENT + return EXTENT; + case 2: // SCAN_ENTRIES + return SCAN_ENTRIES; + 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.EXTENT, new org.apache.thrift.meta_data.FieldMetaData("extent", org.apache.thrift.TFieldRequirementType.DEFAULT, + new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, org.apache.accumulo.core.dataImpl.thrift.TKeyExtent.class))); + tmpMap.put(_Fields.SCAN_ENTRIES, new org.apache.thrift.meta_data.FieldMetaData("scanEntries", org.apache.thrift.TFieldRequirementType.DEFAULT, + new org.apache.thrift.meta_data.ListMetaData(org.apache.thrift.protocol.TType.LIST, + new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)))); + metaDataMap = java.util.Collections.unmodifiableMap(tmpMap); + org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(TTabletRefresh.class, metaDataMap); + } + + public TTabletRefresh() { + } + + public TTabletRefresh( + org.apache.accumulo.core.dataImpl.thrift.TKeyExtent extent, + java.util.List scanEntries) + { + this(); + this.extent = extent; + this.scanEntries = scanEntries; + } + + /** + * Performs a deep copy on other. + */ + public TTabletRefresh(TTabletRefresh other) { + if (other.isSetExtent()) { + this.extent = new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(other.extent); + } + if (other.isSetScanEntries()) { + java.util.List __this__scanEntries = new java.util.ArrayList(other.scanEntries); + this.scanEntries = __this__scanEntries; + } + } + + @Override + public TTabletRefresh deepCopy() { + return new TTabletRefresh(this); + } + + @Override + public void clear() { + this.extent = null; + this.scanEntries = null; + } + + @org.apache.thrift.annotation.Nullable + public org.apache.accumulo.core.dataImpl.thrift.TKeyExtent getExtent() { + return this.extent; + } + + public TTabletRefresh setExtent(@org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TKeyExtent extent) { + this.extent = extent; + return this; + } + + public void unsetExtent() { + this.extent = null; + } + + /** Returns true if field extent is set (has been assigned a value) and false otherwise */ + public boolean isSetExtent() { + return this.extent != null; + } + + public void setExtentIsSet(boolean value) { + if (!value) { + this.extent = null; + } + } + + public int getScanEntriesSize() { + return (this.scanEntries == null) ? 0 : this.scanEntries.size(); + } + + @org.apache.thrift.annotation.Nullable + public java.util.Iterator getScanEntriesIterator() { + return (this.scanEntries == null) ? null : this.scanEntries.iterator(); + } + + public void addToScanEntries(java.lang.String elem) { + if (this.scanEntries == null) { + this.scanEntries = new java.util.ArrayList(); + } + this.scanEntries.add(elem); + } + + @org.apache.thrift.annotation.Nullable + public java.util.List getScanEntries() { + return this.scanEntries; + } + + public TTabletRefresh setScanEntries(@org.apache.thrift.annotation.Nullable java.util.List scanEntries) { + this.scanEntries = scanEntries; + return this; + } + + public void unsetScanEntries() { + this.scanEntries = null; + } + + /** Returns true if field scanEntries is set (has been assigned a value) and false otherwise */ + public boolean isSetScanEntries() { + return this.scanEntries != null; + } + + public void setScanEntriesIsSet(boolean value) { + if (!value) { + this.scanEntries = null; + } + } + + @Override + public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) { + switch (field) { + case EXTENT: + if (value == null) { + unsetExtent(); + } else { + setExtent((org.apache.accumulo.core.dataImpl.thrift.TKeyExtent)value); + } + break; + + case SCAN_ENTRIES: + if (value == null) { + unsetScanEntries(); + } else { + setScanEntries((java.util.List)value); + } + break; + + } + } + + @org.apache.thrift.annotation.Nullable + @Override + public java.lang.Object getFieldValue(_Fields field) { + switch (field) { + case EXTENT: + return getExtent(); + + case SCAN_ENTRIES: + return getScanEntries(); + + } + 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 EXTENT: + return isSetExtent(); + case SCAN_ENTRIES: + return isSetScanEntries(); + } + throw new java.lang.IllegalStateException(); + } + + @Override + public boolean equals(java.lang.Object that) { + if (that instanceof TTabletRefresh) + return this.equals((TTabletRefresh)that); + return false; + } + + public boolean equals(TTabletRefresh that) { + if (that == null) + return false; + if (this == that) + return true; + + boolean this_present_extent = true && this.isSetExtent(); + boolean that_present_extent = true && that.isSetExtent(); + if (this_present_extent || that_present_extent) { + if (!(this_present_extent && that_present_extent)) + return false; + if (!this.extent.equals(that.extent)) + return false; + } + + boolean this_present_scanEntries = true && this.isSetScanEntries(); + boolean that_present_scanEntries = true && that.isSetScanEntries(); + if (this_present_scanEntries || that_present_scanEntries) { + if (!(this_present_scanEntries && that_present_scanEntries)) + return false; + if (!this.scanEntries.equals(that.scanEntries)) + return false; + } + + return true; + } + + @Override + public int hashCode() { + int hashCode = 1; + + hashCode = hashCode * 8191 + ((isSetExtent()) ? 131071 : 524287); + if (isSetExtent()) + hashCode = hashCode * 8191 + extent.hashCode(); + + hashCode = hashCode * 8191 + ((isSetScanEntries()) ? 131071 : 524287); + if (isSetScanEntries()) + hashCode = hashCode * 8191 + scanEntries.hashCode(); + + return hashCode; + } + + @Override + public int compareTo(TTabletRefresh other) { + if (!getClass().equals(other.getClass())) { + return getClass().getName().compareTo(other.getClass().getName()); + } + + int lastComparison = 0; + + lastComparison = java.lang.Boolean.compare(isSetExtent(), other.isSetExtent()); + if (lastComparison != 0) { + return lastComparison; + } + if (isSetExtent()) { + lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.extent, other.extent); + if (lastComparison != 0) { + return lastComparison; + } + } + lastComparison = java.lang.Boolean.compare(isSetScanEntries(), other.isSetScanEntries()); + if (lastComparison != 0) { + return lastComparison; + } + if (isSetScanEntries()) { + lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.scanEntries, other.scanEntries); + 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("TTabletRefresh("); + boolean first = true; + + sb.append("extent:"); + if (this.extent == null) { + sb.append("null"); + } else { + sb.append(this.extent); + } + first = false; + if (!first) sb.append(", "); + sb.append("scanEntries:"); + if (this.scanEntries == null) { + sb.append("null"); + } else { + sb.append(this.scanEntries); + } + 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 (extent != null) { + extent.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 TTabletRefreshStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory { + @Override + public TTabletRefreshStandardScheme getScheme() { + return new TTabletRefreshStandardScheme(); + } + } + + private static class TTabletRefreshStandardScheme extends org.apache.thrift.scheme.StandardScheme { + + @Override + public void read(org.apache.thrift.protocol.TProtocol iprot, TTabletRefresh 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: // EXTENT + if (schemeField.type == org.apache.thrift.protocol.TType.STRUCT) { + struct.extent = new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(); + struct.extent.read(iprot); + struct.setExtentIsSet(true); + } else { + org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); + } + break; + case 2: // SCAN_ENTRIES + if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { + { + org.apache.thrift.protocol.TList _list72 = iprot.readListBegin(); + struct.scanEntries = new java.util.ArrayList(_list72.size); + @org.apache.thrift.annotation.Nullable java.lang.String _elem73; + for (int _i74 = 0; _i74 < _list72.size; ++_i74) + { + _elem73 = iprot.readString(); + struct.scanEntries.add(_elem73); + } + iprot.readListEnd(); + } + struct.setScanEntriesIsSet(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, TTabletRefresh struct) throws org.apache.thrift.TException { + struct.validate(); + + oprot.writeStructBegin(STRUCT_DESC); + if (struct.extent != null) { + oprot.writeFieldBegin(EXTENT_FIELD_DESC); + struct.extent.write(oprot); + oprot.writeFieldEnd(); + } + if (struct.scanEntries != null) { + oprot.writeFieldBegin(SCAN_ENTRIES_FIELD_DESC); + { + oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRING, struct.scanEntries.size())); + for (java.lang.String _iter75 : struct.scanEntries) + { + oprot.writeString(_iter75); + } + oprot.writeListEnd(); + } + oprot.writeFieldEnd(); + } + oprot.writeFieldStop(); + oprot.writeStructEnd(); + } + + } + + private static class TTabletRefreshTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory { + @Override + public TTabletRefreshTupleScheme getScheme() { + return new TTabletRefreshTupleScheme(); + } + } + + private static class TTabletRefreshTupleScheme extends org.apache.thrift.scheme.TupleScheme { + + @Override + public void write(org.apache.thrift.protocol.TProtocol prot, TTabletRefresh 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.isSetExtent()) { + optionals.set(0); + } + if (struct.isSetScanEntries()) { + optionals.set(1); + } + oprot.writeBitSet(optionals, 2); + if (struct.isSetExtent()) { + struct.extent.write(oprot); + } + if (struct.isSetScanEntries()) { + { + oprot.writeI32(struct.scanEntries.size()); + for (java.lang.String _iter76 : struct.scanEntries) + { + oprot.writeString(_iter76); + } + } + } + } + + @Override + public void read(org.apache.thrift.protocol.TProtocol prot, TTabletRefresh 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.extent = new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(); + struct.extent.read(iprot); + struct.setExtentIsSet(true); + } + if (incoming.get(1)) { + { + org.apache.thrift.protocol.TList _list77 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRING); + struct.scanEntries = new java.util.ArrayList(_list77.size); + @org.apache.thrift.annotation.Nullable java.lang.String _elem78; + for (int _i79 = 0; _i79 < _list77.size; ++_i79) + { + _elem78 = iprot.readString(); + struct.scanEntries.add(_elem78); + } + } + struct.setScanEntriesIsSet(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(); + } + private static void unusedMethod() {} +} + diff --git a/core/src/main/thrift-gen-java/org/apache/accumulo/core/tabletserver/thrift/TabletServerClientService.java b/core/src/main/thrift-gen-java/org/apache/accumulo/core/tabletserver/thrift/TabletServerClientService.java index b41c4a0a1a4..42ccb31dff4 100644 --- a/core/src/main/thrift-gen-java/org/apache/accumulo/core/tabletserver/thrift/TabletServerClientService.java +++ b/core/src/main/thrift-gen-java/org/apache/accumulo/core/tabletserver/thrift/TabletServerClientService.java @@ -65,7 +65,7 @@ public interface Iface { public void compactionJobFailed(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.lang.String externalCompactionId, org.apache.accumulo.core.dataImpl.thrift.TKeyExtent extent) throws org.apache.thrift.TException; - public java.util.List refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List extents) throws org.apache.thrift.TException; + public java.util.List refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List tabletsToRefresh) throws org.apache.thrift.TException; } @@ -107,7 +107,7 @@ public interface AsyncIface { public void compactionJobFailed(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.lang.String externalCompactionId, org.apache.accumulo.core.dataImpl.thrift.TKeyExtent extent, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException; - public void refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List extents, org.apache.thrift.async.AsyncMethodCallback> resultHandler) throws org.apache.thrift.TException; + public void refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List tabletsToRefresh, org.apache.thrift.async.AsyncMethodCallback> resultHandler) throws org.apache.thrift.TException; } @@ -579,18 +579,18 @@ public void send_compactionJobFailed(org.apache.accumulo.core.clientImpl.thrift. } @Override - public java.util.List refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List extents) throws org.apache.thrift.TException + public java.util.List refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List tabletsToRefresh) throws org.apache.thrift.TException { - send_refreshTablets(tinfo, credentials, extents); + send_refreshTablets(tinfo, credentials, tabletsToRefresh); return recv_refreshTablets(); } - public void send_refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List extents) throws org.apache.thrift.TException + public void send_refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List tabletsToRefresh) throws org.apache.thrift.TException { refreshTablets_args args = new refreshTablets_args(); args.setTinfo(tinfo); args.setCredentials(credentials); - args.setExtents(extents); + args.setTabletsToRefresh(tabletsToRefresh); sendBase("refreshTablets", args); } @@ -1393,9 +1393,9 @@ public Void getResult() throws org.apache.thrift.TException { } @Override - public void refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List extents, org.apache.thrift.async.AsyncMethodCallback> resultHandler) throws org.apache.thrift.TException { + public void refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List tabletsToRefresh, org.apache.thrift.async.AsyncMethodCallback> resultHandler) throws org.apache.thrift.TException { checkReady(); - refreshTablets_call method_call = new refreshTablets_call(tinfo, credentials, extents, resultHandler, this, ___protocolFactory, ___transport); + refreshTablets_call method_call = new refreshTablets_call(tinfo, credentials, tabletsToRefresh, resultHandler, this, ___protocolFactory, ___transport); this.___currentMethod = method_call; ___manager.call(method_call); } @@ -1403,12 +1403,12 @@ public void refreshTablets(org.apache.accumulo.core.clientImpl.thrift.TInfo tinf public static class refreshTablets_call extends org.apache.thrift.async.TAsyncMethodCall> { private org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo; private org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials; - private java.util.List extents; - public refreshTablets_call(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List extents, 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 { + private java.util.List tabletsToRefresh; + public refreshTablets_call(org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, java.util.List tabletsToRefresh, 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, false); this.tinfo = tinfo; this.credentials = credentials; - this.extents = extents; + this.tabletsToRefresh = tabletsToRefresh; } @Override @@ -1417,7 +1417,7 @@ public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apa refreshTablets_args args = new refreshTablets_args(); args.setTinfo(tinfo); args.setCredentials(credentials); - args.setExtents(extents); + args.setTabletsToRefresh(tabletsToRefresh); args.write(prot); prot.writeMessageEnd(); } @@ -2035,7 +2035,7 @@ protected boolean rethrowUnhandledExceptions() { @Override public refreshTablets_result getResult(I iface, refreshTablets_args args) throws org.apache.thrift.TException { refreshTablets_result result = new refreshTablets_result(); - result.success = iface.refreshTablets(args.tinfo, args.credentials, args.extents); + result.success = iface.refreshTablets(args.tinfo, args.credentials, args.tabletsToRefresh); return result; } } @@ -3229,7 +3229,7 @@ protected boolean isOneway() { @Override public void start(I iface, refreshTablets_args args, org.apache.thrift.async.AsyncMethodCallback> resultHandler) throws org.apache.thrift.TException { - iface.refreshTablets(args.tinfo, args.credentials, args.extents,resultHandler); + iface.refreshTablets(args.tinfo, args.credentials, args.tabletsToRefresh,resultHandler); } } @@ -7109,14 +7109,14 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, getTabletStats_resu case 0: // SUCCESS if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { { - org.apache.thrift.protocol.TList _list72 = iprot.readListBegin(); - struct.success = new java.util.ArrayList(_list72.size); - @org.apache.thrift.annotation.Nullable TabletStats _elem73; - for (int _i74 = 0; _i74 < _list72.size; ++_i74) + org.apache.thrift.protocol.TList _list80 = iprot.readListBegin(); + struct.success = new java.util.ArrayList(_list80.size); + @org.apache.thrift.annotation.Nullable TabletStats _elem81; + for (int _i82 = 0; _i82 < _list80.size; ++_i82) { - _elem73 = new TabletStats(); - _elem73.read(iprot); - struct.success.add(_elem73); + _elem81 = new TabletStats(); + _elem81.read(iprot); + struct.success.add(_elem81); } iprot.readListEnd(); } @@ -7154,9 +7154,9 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, getTabletStats_res oprot.writeFieldBegin(SUCCESS_FIELD_DESC); { oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.success.size())); - for (TabletStats _iter75 : struct.success) + for (TabletStats _iter83 : struct.success) { - _iter75.write(oprot); + _iter83.write(oprot); } oprot.writeListEnd(); } @@ -7196,9 +7196,9 @@ public void write(org.apache.thrift.protocol.TProtocol prot, getTabletStats_resu if (struct.isSetSuccess()) { { oprot.writeI32(struct.success.size()); - for (TabletStats _iter76 : struct.success) + for (TabletStats _iter84 : struct.success) { - _iter76.write(oprot); + _iter84.write(oprot); } } } @@ -7213,14 +7213,14 @@ public void read(org.apache.thrift.protocol.TProtocol prot, getTabletStats_resul java.util.BitSet incoming = iprot.readBitSet(2); if (incoming.get(0)) { { - org.apache.thrift.protocol.TList _list77 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); - struct.success = new java.util.ArrayList(_list77.size); - @org.apache.thrift.annotation.Nullable TabletStats _elem78; - for (int _i79 = 0; _i79 < _list77.size; ++_i79) + org.apache.thrift.protocol.TList _list85 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); + struct.success = new java.util.ArrayList(_list85.size); + @org.apache.thrift.annotation.Nullable TabletStats _elem86; + for (int _i87 = 0; _i87 < _list85.size; ++_i87) { - _elem78 = new TabletStats(); - _elem78.read(iprot); - struct.success.add(_elem78); + _elem86 = new TabletStats(); + _elem86.read(iprot); + struct.success.add(_elem86); } } struct.setSuccessIsSet(true); @@ -10705,14 +10705,14 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, getActiveCompaction case 0: // SUCCESS if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { { - org.apache.thrift.protocol.TList _list80 = iprot.readListBegin(); - struct.success = new java.util.ArrayList(_list80.size); - @org.apache.thrift.annotation.Nullable ActiveCompaction _elem81; - for (int _i82 = 0; _i82 < _list80.size; ++_i82) + org.apache.thrift.protocol.TList _list88 = iprot.readListBegin(); + struct.success = new java.util.ArrayList(_list88.size); + @org.apache.thrift.annotation.Nullable ActiveCompaction _elem89; + for (int _i90 = 0; _i90 < _list88.size; ++_i90) { - _elem81 = new ActiveCompaction(); - _elem81.read(iprot); - struct.success.add(_elem81); + _elem89 = new ActiveCompaction(); + _elem89.read(iprot); + struct.success.add(_elem89); } iprot.readListEnd(); } @@ -10750,9 +10750,9 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, getActiveCompactio oprot.writeFieldBegin(SUCCESS_FIELD_DESC); { oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.success.size())); - for (ActiveCompaction _iter83 : struct.success) + for (ActiveCompaction _iter91 : struct.success) { - _iter83.write(oprot); + _iter91.write(oprot); } oprot.writeListEnd(); } @@ -10792,9 +10792,9 @@ public void write(org.apache.thrift.protocol.TProtocol prot, getActiveCompaction if (struct.isSetSuccess()) { { oprot.writeI32(struct.success.size()); - for (ActiveCompaction _iter84 : struct.success) + for (ActiveCompaction _iter92 : struct.success) { - _iter84.write(oprot); + _iter92.write(oprot); } } } @@ -10809,14 +10809,14 @@ public void read(org.apache.thrift.protocol.TProtocol prot, getActiveCompactions java.util.BitSet incoming = iprot.readBitSet(2); if (incoming.get(0)) { { - org.apache.thrift.protocol.TList _list85 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); - struct.success = new java.util.ArrayList(_list85.size); - @org.apache.thrift.annotation.Nullable ActiveCompaction _elem86; - for (int _i87 = 0; _i87 < _list85.size; ++_i87) + org.apache.thrift.protocol.TList _list93 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); + struct.success = new java.util.ArrayList(_list93.size); + @org.apache.thrift.annotation.Nullable ActiveCompaction _elem94; + for (int _i95 = 0; _i95 < _list93.size; ++_i95) { - _elem86 = new ActiveCompaction(); - _elem86.read(iprot); - struct.success.add(_elem86); + _elem94 = new ActiveCompaction(); + _elem94.read(iprot); + struct.success.add(_elem94); } } struct.setSuccessIsSet(true); @@ -11349,13 +11349,13 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, removeLogs_args str case 3: // FILENAMES if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { { - org.apache.thrift.protocol.TList _list88 = iprot.readListBegin(); - struct.filenames = new java.util.ArrayList(_list88.size); - @org.apache.thrift.annotation.Nullable java.lang.String _elem89; - for (int _i90 = 0; _i90 < _list88.size; ++_i90) + org.apache.thrift.protocol.TList _list96 = iprot.readListBegin(); + struct.filenames = new java.util.ArrayList(_list96.size); + @org.apache.thrift.annotation.Nullable java.lang.String _elem97; + for (int _i98 = 0; _i98 < _list96.size; ++_i98) { - _elem89 = iprot.readString(); - struct.filenames.add(_elem89); + _elem97 = iprot.readString(); + struct.filenames.add(_elem97); } iprot.readListEnd(); } @@ -11394,9 +11394,9 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, removeLogs_args st oprot.writeFieldBegin(FILENAMES_FIELD_DESC); { oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRING, struct.filenames.size())); - for (java.lang.String _iter91 : struct.filenames) + for (java.lang.String _iter99 : struct.filenames) { - oprot.writeString(_iter91); + oprot.writeString(_iter99); } oprot.writeListEnd(); } @@ -11440,9 +11440,9 @@ public void write(org.apache.thrift.protocol.TProtocol prot, removeLogs_args str if (struct.isSetFilenames()) { { oprot.writeI32(struct.filenames.size()); - for (java.lang.String _iter92 : struct.filenames) + for (java.lang.String _iter100 : struct.filenames) { - oprot.writeString(_iter92); + oprot.writeString(_iter100); } } } @@ -11464,13 +11464,13 @@ public void read(org.apache.thrift.protocol.TProtocol prot, removeLogs_args stru } if (incoming.get(2)) { { - org.apache.thrift.protocol.TList _list93 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRING); - struct.filenames = new java.util.ArrayList(_list93.size); - @org.apache.thrift.annotation.Nullable java.lang.String _elem94; - for (int _i95 = 0; _i95 < _list93.size; ++_i95) + org.apache.thrift.protocol.TList _list101 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRING); + struct.filenames = new java.util.ArrayList(_list101.size); + @org.apache.thrift.annotation.Nullable java.lang.String _elem102; + for (int _i103 = 0; _i103 < _list101.size; ++_i103) { - _elem94 = iprot.readString(); - struct.filenames.add(_elem94); + _elem102 = iprot.readString(); + struct.filenames.add(_elem102); } } struct.setFilenamesIsSet(true); @@ -12303,13 +12303,13 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, getActiveLogs_resul case 0: // SUCCESS if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { { - org.apache.thrift.protocol.TList _list96 = iprot.readListBegin(); - struct.success = new java.util.ArrayList(_list96.size); - @org.apache.thrift.annotation.Nullable java.lang.String _elem97; - for (int _i98 = 0; _i98 < _list96.size; ++_i98) + org.apache.thrift.protocol.TList _list104 = iprot.readListBegin(); + struct.success = new java.util.ArrayList(_list104.size); + @org.apache.thrift.annotation.Nullable java.lang.String _elem105; + for (int _i106 = 0; _i106 < _list104.size; ++_i106) { - _elem97 = iprot.readString(); - struct.success.add(_elem97); + _elem105 = iprot.readString(); + struct.success.add(_elem105); } iprot.readListEnd(); } @@ -12338,9 +12338,9 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, getActiveLogs_resu oprot.writeFieldBegin(SUCCESS_FIELD_DESC); { oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRING, struct.success.size())); - for (java.lang.String _iter99 : struct.success) + for (java.lang.String _iter107 : struct.success) { - oprot.writeString(_iter99); + oprot.writeString(_iter107); } oprot.writeListEnd(); } @@ -12372,9 +12372,9 @@ public void write(org.apache.thrift.protocol.TProtocol prot, getActiveLogs_resul if (struct.isSetSuccess()) { { oprot.writeI32(struct.success.size()); - for (java.lang.String _iter100 : struct.success) + for (java.lang.String _iter108 : struct.success) { - oprot.writeString(_iter100); + oprot.writeString(_iter108); } } } @@ -12386,13 +12386,13 @@ public void read(org.apache.thrift.protocol.TProtocol prot, getActiveLogs_result java.util.BitSet incoming = iprot.readBitSet(1); if (incoming.get(0)) { { - org.apache.thrift.protocol.TList _list101 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRING); - struct.success = new java.util.ArrayList(_list101.size); - @org.apache.thrift.annotation.Nullable java.lang.String _elem102; - for (int _i103 = 0; _i103 < _list101.size; ++_i103) + org.apache.thrift.protocol.TList _list109 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRING); + struct.success = new java.util.ArrayList(_list109.size); + @org.apache.thrift.annotation.Nullable java.lang.String _elem110; + for (int _i111 = 0; _i111 < _list109.size; ++_i111) { - _elem102 = iprot.readString(); - struct.success.add(_elem102); + _elem110 = iprot.readString(); + struct.success.add(_elem110); } } struct.setSuccessIsSet(true); @@ -15516,26 +15516,26 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, startGetSummariesFr case 4: // FILES if (schemeField.type == org.apache.thrift.protocol.TType.MAP) { { - org.apache.thrift.protocol.TMap _map104 = iprot.readMapBegin(); - struct.files = new java.util.HashMap>(2*_map104.size); - @org.apache.thrift.annotation.Nullable java.lang.String _key105; - @org.apache.thrift.annotation.Nullable java.util.List _val106; - for (int _i107 = 0; _i107 < _map104.size; ++_i107) + org.apache.thrift.protocol.TMap _map112 = iprot.readMapBegin(); + struct.files = new java.util.HashMap>(2*_map112.size); + @org.apache.thrift.annotation.Nullable java.lang.String _key113; + @org.apache.thrift.annotation.Nullable java.util.List _val114; + for (int _i115 = 0; _i115 < _map112.size; ++_i115) { - _key105 = iprot.readString(); + _key113 = iprot.readString(); { - org.apache.thrift.protocol.TList _list108 = iprot.readListBegin(); - _val106 = new java.util.ArrayList(_list108.size); - @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TRowRange _elem109; - for (int _i110 = 0; _i110 < _list108.size; ++_i110) + org.apache.thrift.protocol.TList _list116 = iprot.readListBegin(); + _val114 = new java.util.ArrayList(_list116.size); + @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TRowRange _elem117; + for (int _i118 = 0; _i118 < _list116.size; ++_i118) { - _elem109 = new org.apache.accumulo.core.dataImpl.thrift.TRowRange(); - _elem109.read(iprot); - _val106.add(_elem109); + _elem117 = new org.apache.accumulo.core.dataImpl.thrift.TRowRange(); + _elem117.read(iprot); + _val114.add(_elem117); } iprot.readListEnd(); } - struct.files.put(_key105, _val106); + struct.files.put(_key113, _val114); } iprot.readMapEnd(); } @@ -15579,14 +15579,14 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, startGetSummariesF oprot.writeFieldBegin(FILES_FIELD_DESC); { oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.LIST, struct.files.size())); - for (java.util.Map.Entry> _iter111 : struct.files.entrySet()) + for (java.util.Map.Entry> _iter119 : struct.files.entrySet()) { - oprot.writeString(_iter111.getKey()); + oprot.writeString(_iter119.getKey()); { - oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, _iter111.getValue().size())); - for (org.apache.accumulo.core.dataImpl.thrift.TRowRange _iter112 : _iter111.getValue()) + oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, _iter119.getValue().size())); + for (org.apache.accumulo.core.dataImpl.thrift.TRowRange _iter120 : _iter119.getValue()) { - _iter112.write(oprot); + _iter120.write(oprot); } oprot.writeListEnd(); } @@ -15639,14 +15639,14 @@ public void write(org.apache.thrift.protocol.TProtocol prot, startGetSummariesFr if (struct.isSetFiles()) { { oprot.writeI32(struct.files.size()); - for (java.util.Map.Entry> _iter113 : struct.files.entrySet()) + for (java.util.Map.Entry> _iter121 : struct.files.entrySet()) { - oprot.writeString(_iter113.getKey()); + oprot.writeString(_iter121.getKey()); { - oprot.writeI32(_iter113.getValue().size()); - for (org.apache.accumulo.core.dataImpl.thrift.TRowRange _iter114 : _iter113.getValue()) + oprot.writeI32(_iter121.getValue().size()); + for (org.apache.accumulo.core.dataImpl.thrift.TRowRange _iter122 : _iter121.getValue()) { - _iter114.write(oprot); + _iter122.write(oprot); } } } @@ -15675,25 +15675,25 @@ public void read(org.apache.thrift.protocol.TProtocol prot, startGetSummariesFro } if (incoming.get(3)) { { - org.apache.thrift.protocol.TMap _map115 = iprot.readMapBegin(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.LIST); - struct.files = new java.util.HashMap>(2*_map115.size); - @org.apache.thrift.annotation.Nullable java.lang.String _key116; - @org.apache.thrift.annotation.Nullable java.util.List _val117; - for (int _i118 = 0; _i118 < _map115.size; ++_i118) + org.apache.thrift.protocol.TMap _map123 = iprot.readMapBegin(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.LIST); + struct.files = new java.util.HashMap>(2*_map123.size); + @org.apache.thrift.annotation.Nullable java.lang.String _key124; + @org.apache.thrift.annotation.Nullable java.util.List _val125; + for (int _i126 = 0; _i126 < _map123.size; ++_i126) { - _key116 = iprot.readString(); + _key124 = iprot.readString(); { - org.apache.thrift.protocol.TList _list119 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); - _val117 = new java.util.ArrayList(_list119.size); - @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TRowRange _elem120; - for (int _i121 = 0; _i121 < _list119.size; ++_i121) + org.apache.thrift.protocol.TList _list127 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); + _val125 = new java.util.ArrayList(_list127.size); + @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TRowRange _elem128; + for (int _i129 = 0; _i129 < _list127.size; ++_i129) { - _elem120 = new org.apache.accumulo.core.dataImpl.thrift.TRowRange(); - _elem120.read(iprot); - _val117.add(_elem120); + _elem128 = new org.apache.accumulo.core.dataImpl.thrift.TRowRange(); + _elem128.read(iprot); + _val125.add(_elem128); } } - struct.files.put(_key116, _val117); + struct.files.put(_key124, _val125); } } struct.setFilesIsSet(true); @@ -18075,14 +18075,14 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, getCompactionQueueI case 0: // SUCCESS if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { { - org.apache.thrift.protocol.TList _list122 = iprot.readListBegin(); - struct.success = new java.util.ArrayList(_list122.size); - @org.apache.thrift.annotation.Nullable TCompactionQueueSummary _elem123; - for (int _i124 = 0; _i124 < _list122.size; ++_i124) + org.apache.thrift.protocol.TList _list130 = iprot.readListBegin(); + struct.success = new java.util.ArrayList(_list130.size); + @org.apache.thrift.annotation.Nullable TCompactionQueueSummary _elem131; + for (int _i132 = 0; _i132 < _list130.size; ++_i132) { - _elem123 = new TCompactionQueueSummary(); - _elem123.read(iprot); - struct.success.add(_elem123); + _elem131 = new TCompactionQueueSummary(); + _elem131.read(iprot); + struct.success.add(_elem131); } iprot.readListEnd(); } @@ -18120,9 +18120,9 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, getCompactionQueue oprot.writeFieldBegin(SUCCESS_FIELD_DESC); { oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.success.size())); - for (TCompactionQueueSummary _iter125 : struct.success) + for (TCompactionQueueSummary _iter133 : struct.success) { - _iter125.write(oprot); + _iter133.write(oprot); } oprot.writeListEnd(); } @@ -18162,9 +18162,9 @@ public void write(org.apache.thrift.protocol.TProtocol prot, getCompactionQueueI if (struct.isSetSuccess()) { { oprot.writeI32(struct.success.size()); - for (TCompactionQueueSummary _iter126 : struct.success) + for (TCompactionQueueSummary _iter134 : struct.success) { - _iter126.write(oprot); + _iter134.write(oprot); } } } @@ -18179,14 +18179,14 @@ public void read(org.apache.thrift.protocol.TProtocol prot, getCompactionQueueIn java.util.BitSet incoming = iprot.readBitSet(2); if (incoming.get(0)) { { - org.apache.thrift.protocol.TList _list127 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); - struct.success = new java.util.ArrayList(_list127.size); - @org.apache.thrift.annotation.Nullable TCompactionQueueSummary _elem128; - for (int _i129 = 0; _i129 < _list127.size; ++_i129) + org.apache.thrift.protocol.TList _list135 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); + struct.success = new java.util.ArrayList(_list135.size); + @org.apache.thrift.annotation.Nullable TCompactionQueueSummary _elem136; + for (int _i137 = 0; _i137 < _list135.size; ++_i137) { - _elem128 = new TCompactionQueueSummary(); - _elem128.read(iprot); - struct.success.add(_elem128); + _elem136 = new TCompactionQueueSummary(); + _elem136.read(iprot); + struct.success.add(_elem136); } } struct.setSuccessIsSet(true); @@ -21223,20 +21223,20 @@ public static class refreshTablets_args implements org.apache.thrift.TBase extents; // required + public @org.apache.thrift.annotation.Nullable java.util.List tabletsToRefresh; // 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"), - EXTENTS((short)3, "extents"); + TABLETS_TO_REFRESH((short)3, "tabletsToRefresh"); private static final java.util.Map byName = new java.util.HashMap(); @@ -21256,8 +21256,8 @@ public static _Fields findByThriftId(int fieldId) { return TINFO; case 2: // CREDENTIALS return CREDENTIALS; - case 3: // EXTENTS - return EXTENTS; + case 3: // TABLETS_TO_REFRESH + return TABLETS_TO_REFRESH; default: return null; } @@ -21308,9 +21308,9 @@ public java.lang.String getFieldName() { new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, org.apache.accumulo.core.clientImpl.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))); - tmpMap.put(_Fields.EXTENTS, new org.apache.thrift.meta_data.FieldMetaData("extents", org.apache.thrift.TFieldRequirementType.DEFAULT, + tmpMap.put(_Fields.TABLETS_TO_REFRESH, new org.apache.thrift.meta_data.FieldMetaData("tabletsToRefresh", org.apache.thrift.TFieldRequirementType.DEFAULT, new org.apache.thrift.meta_data.ListMetaData(org.apache.thrift.protocol.TType.LIST, - new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, org.apache.accumulo.core.dataImpl.thrift.TKeyExtent.class)))); + new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, TTabletRefresh.class)))); metaDataMap = java.util.Collections.unmodifiableMap(tmpMap); org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(refreshTablets_args.class, metaDataMap); } @@ -21321,12 +21321,12 @@ public refreshTablets_args() { public refreshTablets_args( org.apache.accumulo.core.clientImpl.thrift.TInfo tinfo, org.apache.accumulo.core.securityImpl.thrift.TCredentials credentials, - java.util.List extents) + java.util.List tabletsToRefresh) { this(); this.tinfo = tinfo; this.credentials = credentials; - this.extents = extents; + this.tabletsToRefresh = tabletsToRefresh; } /** @@ -21339,12 +21339,12 @@ public refreshTablets_args(refreshTablets_args other) { if (other.isSetCredentials()) { this.credentials = new org.apache.accumulo.core.securityImpl.thrift.TCredentials(other.credentials); } - if (other.isSetExtents()) { - java.util.List __this__extents = new java.util.ArrayList(other.extents.size()); - for (org.apache.accumulo.core.dataImpl.thrift.TKeyExtent other_element : other.extents) { - __this__extents.add(new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(other_element)); + if (other.isSetTabletsToRefresh()) { + java.util.List __this__tabletsToRefresh = new java.util.ArrayList(other.tabletsToRefresh.size()); + for (TTabletRefresh other_element : other.tabletsToRefresh) { + __this__tabletsToRefresh.add(new TTabletRefresh(other_element)); } - this.extents = __this__extents; + this.tabletsToRefresh = __this__tabletsToRefresh; } } @@ -21357,7 +21357,7 @@ public refreshTablets_args deepCopy() { public void clear() { this.tinfo = null; this.credentials = null; - this.extents = null; + this.tabletsToRefresh = null; } @org.apache.thrift.annotation.Nullable @@ -21410,44 +21410,44 @@ public void setCredentialsIsSet(boolean value) { } } - public int getExtentsSize() { - return (this.extents == null) ? 0 : this.extents.size(); + public int getTabletsToRefreshSize() { + return (this.tabletsToRefresh == null) ? 0 : this.tabletsToRefresh.size(); } @org.apache.thrift.annotation.Nullable - public java.util.Iterator getExtentsIterator() { - return (this.extents == null) ? null : this.extents.iterator(); + public java.util.Iterator getTabletsToRefreshIterator() { + return (this.tabletsToRefresh == null) ? null : this.tabletsToRefresh.iterator(); } - public void addToExtents(org.apache.accumulo.core.dataImpl.thrift.TKeyExtent elem) { - if (this.extents == null) { - this.extents = new java.util.ArrayList(); + public void addToTabletsToRefresh(TTabletRefresh elem) { + if (this.tabletsToRefresh == null) { + this.tabletsToRefresh = new java.util.ArrayList(); } - this.extents.add(elem); + this.tabletsToRefresh.add(elem); } @org.apache.thrift.annotation.Nullable - public java.util.List getExtents() { - return this.extents; + public java.util.List getTabletsToRefresh() { + return this.tabletsToRefresh; } - public refreshTablets_args setExtents(@org.apache.thrift.annotation.Nullable java.util.List extents) { - this.extents = extents; + public refreshTablets_args setTabletsToRefresh(@org.apache.thrift.annotation.Nullable java.util.List tabletsToRefresh) { + this.tabletsToRefresh = tabletsToRefresh; return this; } - public void unsetExtents() { - this.extents = null; + public void unsetTabletsToRefresh() { + this.tabletsToRefresh = null; } - /** Returns true if field extents is set (has been assigned a value) and false otherwise */ - public boolean isSetExtents() { - return this.extents != null; + /** Returns true if field tabletsToRefresh is set (has been assigned a value) and false otherwise */ + public boolean isSetTabletsToRefresh() { + return this.tabletsToRefresh != null; } - public void setExtentsIsSet(boolean value) { + public void setTabletsToRefreshIsSet(boolean value) { if (!value) { - this.extents = null; + this.tabletsToRefresh = null; } } @@ -21470,11 +21470,11 @@ public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable } break; - case EXTENTS: + case TABLETS_TO_REFRESH: if (value == null) { - unsetExtents(); + unsetTabletsToRefresh(); } else { - setExtents((java.util.List)value); + setTabletsToRefresh((java.util.List)value); } break; @@ -21491,8 +21491,8 @@ public java.lang.Object getFieldValue(_Fields field) { case CREDENTIALS: return getCredentials(); - case EXTENTS: - return getExtents(); + case TABLETS_TO_REFRESH: + return getTabletsToRefresh(); } throw new java.lang.IllegalStateException(); @@ -21510,8 +21510,8 @@ public boolean isSet(_Fields field) { return isSetTinfo(); case CREDENTIALS: return isSetCredentials(); - case EXTENTS: - return isSetExtents(); + case TABLETS_TO_REFRESH: + return isSetTabletsToRefresh(); } throw new java.lang.IllegalStateException(); } @@ -21547,12 +21547,12 @@ public boolean equals(refreshTablets_args that) { return false; } - boolean this_present_extents = true && this.isSetExtents(); - boolean that_present_extents = true && that.isSetExtents(); - if (this_present_extents || that_present_extents) { - if (!(this_present_extents && that_present_extents)) + boolean this_present_tabletsToRefresh = true && this.isSetTabletsToRefresh(); + boolean that_present_tabletsToRefresh = true && that.isSetTabletsToRefresh(); + if (this_present_tabletsToRefresh || that_present_tabletsToRefresh) { + if (!(this_present_tabletsToRefresh && that_present_tabletsToRefresh)) return false; - if (!this.extents.equals(that.extents)) + if (!this.tabletsToRefresh.equals(that.tabletsToRefresh)) return false; } @@ -21571,9 +21571,9 @@ public int hashCode() { if (isSetCredentials()) hashCode = hashCode * 8191 + credentials.hashCode(); - hashCode = hashCode * 8191 + ((isSetExtents()) ? 131071 : 524287); - if (isSetExtents()) - hashCode = hashCode * 8191 + extents.hashCode(); + hashCode = hashCode * 8191 + ((isSetTabletsToRefresh()) ? 131071 : 524287); + if (isSetTabletsToRefresh()) + hashCode = hashCode * 8191 + tabletsToRefresh.hashCode(); return hashCode; } @@ -21606,12 +21606,12 @@ public int compareTo(refreshTablets_args other) { return lastComparison; } } - lastComparison = java.lang.Boolean.compare(isSetExtents(), other.isSetExtents()); + lastComparison = java.lang.Boolean.compare(isSetTabletsToRefresh(), other.isSetTabletsToRefresh()); if (lastComparison != 0) { return lastComparison; } - if (isSetExtents()) { - lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.extents, other.extents); + if (isSetTabletsToRefresh()) { + lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.tabletsToRefresh, other.tabletsToRefresh); if (lastComparison != 0) { return lastComparison; } @@ -21656,11 +21656,11 @@ public java.lang.String toString() { } first = false; if (!first) sb.append(", "); - sb.append("extents:"); - if (this.extents == null) { + sb.append("tabletsToRefresh:"); + if (this.tabletsToRefresh == null) { sb.append("null"); } else { - sb.append(this.extents); + sb.append(this.tabletsToRefresh); } first = false; sb.append(")"); @@ -21732,21 +21732,21 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, refreshTablets_args org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); } break; - case 3: // EXTENTS + case 3: // TABLETS_TO_REFRESH if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { { - org.apache.thrift.protocol.TList _list130 = iprot.readListBegin(); - struct.extents = new java.util.ArrayList(_list130.size); - @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _elem131; - for (int _i132 = 0; _i132 < _list130.size; ++_i132) + org.apache.thrift.protocol.TList _list138 = iprot.readListBegin(); + struct.tabletsToRefresh = new java.util.ArrayList(_list138.size); + @org.apache.thrift.annotation.Nullable TTabletRefresh _elem139; + for (int _i140 = 0; _i140 < _list138.size; ++_i140) { - _elem131 = new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(); - _elem131.read(iprot); - struct.extents.add(_elem131); + _elem139 = new TTabletRefresh(); + _elem139.read(iprot); + struct.tabletsToRefresh.add(_elem139); } iprot.readListEnd(); } - struct.setExtentsIsSet(true); + struct.setTabletsToRefreshIsSet(true); } else { org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); } @@ -21777,13 +21777,13 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, refreshTablets_arg struct.credentials.write(oprot); oprot.writeFieldEnd(); } - if (struct.extents != null) { - oprot.writeFieldBegin(EXTENTS_FIELD_DESC); + if (struct.tabletsToRefresh != null) { + oprot.writeFieldBegin(TABLETS_TO_REFRESH_FIELD_DESC); { - oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.extents.size())); - for (org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _iter133 : struct.extents) + oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.tabletsToRefresh.size())); + for (TTabletRefresh _iter141 : struct.tabletsToRefresh) { - _iter133.write(oprot); + _iter141.write(oprot); } oprot.writeListEnd(); } @@ -21814,7 +21814,7 @@ public void write(org.apache.thrift.protocol.TProtocol prot, refreshTablets_args if (struct.isSetCredentials()) { optionals.set(1); } - if (struct.isSetExtents()) { + if (struct.isSetTabletsToRefresh()) { optionals.set(2); } oprot.writeBitSet(optionals, 3); @@ -21824,12 +21824,12 @@ public void write(org.apache.thrift.protocol.TProtocol prot, refreshTablets_args if (struct.isSetCredentials()) { struct.credentials.write(oprot); } - if (struct.isSetExtents()) { + if (struct.isSetTabletsToRefresh()) { { - oprot.writeI32(struct.extents.size()); - for (org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _iter134 : struct.extents) + oprot.writeI32(struct.tabletsToRefresh.size()); + for (TTabletRefresh _iter142 : struct.tabletsToRefresh) { - _iter134.write(oprot); + _iter142.write(oprot); } } } @@ -21851,17 +21851,17 @@ public void read(org.apache.thrift.protocol.TProtocol prot, refreshTablets_args } if (incoming.get(2)) { { - org.apache.thrift.protocol.TList _list135 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); - struct.extents = new java.util.ArrayList(_list135.size); - @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _elem136; - for (int _i137 = 0; _i137 < _list135.size; ++_i137) + org.apache.thrift.protocol.TList _list143 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); + struct.tabletsToRefresh = new java.util.ArrayList(_list143.size); + @org.apache.thrift.annotation.Nullable TTabletRefresh _elem144; + for (int _i145 = 0; _i145 < _list143.size; ++_i145) { - _elem136 = new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(); - _elem136.read(iprot); - struct.extents.add(_elem136); + _elem144 = new TTabletRefresh(); + _elem144.read(iprot); + struct.tabletsToRefresh.add(_elem144); } } - struct.setExtentsIsSet(true); + struct.setTabletsToRefreshIsSet(true); } } } @@ -22200,14 +22200,14 @@ public void read(org.apache.thrift.protocol.TProtocol iprot, refreshTablets_resu case 0: // SUCCESS if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { { - org.apache.thrift.protocol.TList _list138 = iprot.readListBegin(); - struct.success = new java.util.ArrayList(_list138.size); - @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _elem139; - for (int _i140 = 0; _i140 < _list138.size; ++_i140) + org.apache.thrift.protocol.TList _list146 = iprot.readListBegin(); + struct.success = new java.util.ArrayList(_list146.size); + @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _elem147; + for (int _i148 = 0; _i148 < _list146.size; ++_i148) { - _elem139 = new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(); - _elem139.read(iprot); - struct.success.add(_elem139); + _elem147 = new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(); + _elem147.read(iprot); + struct.success.add(_elem147); } iprot.readListEnd(); } @@ -22236,9 +22236,9 @@ public void write(org.apache.thrift.protocol.TProtocol oprot, refreshTablets_res oprot.writeFieldBegin(SUCCESS_FIELD_DESC); { oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.success.size())); - for (org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _iter141 : struct.success) + for (org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _iter149 : struct.success) { - _iter141.write(oprot); + _iter149.write(oprot); } oprot.writeListEnd(); } @@ -22270,9 +22270,9 @@ public void write(org.apache.thrift.protocol.TProtocol prot, refreshTablets_resu if (struct.isSetSuccess()) { { oprot.writeI32(struct.success.size()); - for (org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _iter142 : struct.success) + for (org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _iter150 : struct.success) { - _iter142.write(oprot); + _iter150.write(oprot); } } } @@ -22284,14 +22284,14 @@ public void read(org.apache.thrift.protocol.TProtocol prot, refreshTablets_resul java.util.BitSet incoming = iprot.readBitSet(1); if (incoming.get(0)) { { - org.apache.thrift.protocol.TList _list143 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); - struct.success = new java.util.ArrayList(_list143.size); - @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _elem144; - for (int _i145 = 0; _i145 < _list143.size; ++_i145) + org.apache.thrift.protocol.TList _list151 = iprot.readListBegin(org.apache.thrift.protocol.TType.STRUCT); + struct.success = new java.util.ArrayList(_list151.size); + @org.apache.thrift.annotation.Nullable org.apache.accumulo.core.dataImpl.thrift.TKeyExtent _elem152; + for (int _i153 = 0; _i153 < _list151.size; ++_i153) { - _elem144 = new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(); - _elem144.read(iprot); - struct.success.add(_elem144); + _elem152 = new org.apache.accumulo.core.dataImpl.thrift.TKeyExtent(); + _elem152.read(iprot); + struct.success.add(_elem152); } } struct.setSuccessIsSet(true); diff --git a/core/src/main/thrift/tabletserver.thrift b/core/src/main/thrift/tabletserver.thrift index 06b299ac132..87c0e19acc8 100644 --- a/core/src/main/thrift/tabletserver.thrift +++ b/core/src/main/thrift/tabletserver.thrift @@ -133,6 +133,11 @@ struct TCompactionStats{ 3:i64 fileSize; } +struct TTabletRefresh { + 1:data.TKeyExtent extent + 2:list scanEntries +} + service TabletServerClientService { oneway void flush( @@ -279,7 +284,7 @@ service TabletServerClientService { list refreshTablets( 1:client.TInfo tinfo 2:security.TCredentials credentials - 3:list extents + 3:list tabletsToRefresh ) } diff --git a/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java b/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java index 94250b11c42..44f04294a71 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java +++ b/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java @@ -45,6 +45,7 @@ import org.apache.accumulo.server.conf.codec.VersionedProperties; import org.apache.accumulo.server.conf.store.SystemPropKey; import org.apache.accumulo.server.log.WalStateManager; +import org.apache.accumulo.server.metadata.RefreshesImpl; import org.apache.accumulo.server.metadata.RootGcCandidates; import org.apache.accumulo.server.tables.TableManager; import org.apache.zookeeper.KeeperException; @@ -132,6 +133,8 @@ void initialize(final ServerContext context, final boolean clearInstanceName, ZooUtil.NodeExistsPolicy.FAIL); zoo.putPersistentData(zkInstanceRoot + RootTable.ZROOT_TABLET_GC_CANDIDATES, new RootGcCandidates().toJson().getBytes(UTF_8), ZooUtil.NodeExistsPolicy.FAIL); + zoo.putPersistentData(zkInstanceRoot + RootTable.ZROOT_TABLET_REFRESHES, + RefreshesImpl.getInitialJson().getBytes(UTF_8), ZooUtil.NodeExistsPolicy.FAIL); zoo.putPersistentData(zkInstanceRoot + Constants.ZMANAGERS, EMPTY_BYTE_ARRAY, ZooUtil.NodeExistsPolicy.FAIL); zoo.putPersistentData(zkInstanceRoot + Constants.ZMANAGER_LOCK, EMPTY_BYTE_ARRAY, diff --git a/server/base/src/main/java/org/apache/accumulo/server/metadata/RefreshesImpl.java b/server/base/src/main/java/org/apache/accumulo/server/metadata/RefreshesImpl.java new file mode 100644 index 00000000000..531136853b4 --- /dev/null +++ b/server/base/src/main/java/org/apache/accumulo/server/metadata/RefreshesImpl.java @@ -0,0 +1,254 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.metadata; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.accumulo.core.metadata.RootTable.ZROOT_TABLET_REFRESHES; +import static org.apache.accumulo.core.util.LazySingletons.GSON; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Collection; +import java.util.Map; +import java.util.Objects; +import java.util.function.Consumer; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import org.apache.accumulo.core.client.BatchWriter; +import org.apache.accumulo.core.client.MutationsRejectedException; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.client.TableNotFoundException; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Mutation; +import org.apache.accumulo.core.data.Range; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.metadata.RootTable; +import org.apache.accumulo.core.metadata.TServerInstance; +import org.apache.accumulo.core.metadata.schema.Ample; +import org.apache.accumulo.core.metadata.schema.ExternalCompactionId; +import org.apache.accumulo.core.metadata.schema.MetadataSchema.RefreshSection; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.accumulo.server.ServerContext; +import org.apache.hadoop.io.Text; +import org.apache.zookeeper.KeeperException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Preconditions; + +public class RefreshesImpl implements Ample.Refreshes { + + private static final Logger log = LoggerFactory.getLogger(RefreshesImpl.class); + + private final Ample.DataLevel dataLevel; + private final ServerContext context; + + public static String getInitialJson() { + return toJson(Map.of()); + } + + // the expected version of serialized data + private static final int CURRENT_VERSION = 1; + + // This class is used to serialize and deserialize root tablet metadata using GSon. Any changes to + // this class must consider persisted data. + private static class Data { + + final int version; + + final Map entries; + + public Data(int version, Map entries) { + this.version = version; + this.entries = entries; + } + } + + private static String toJson(Map entries) { + var data = new Data(CURRENT_VERSION, Objects.requireNonNull(entries)); + return GSON.get().toJson(data, Data.class); + } + + private Map fromJson(String json) { + Data data = GSON.get().fromJson(json, Data.class); + Preconditions.checkArgument(data.version == CURRENT_VERSION, "Expected version %s saw %s", + CURRENT_VERSION, data.version); + Objects.requireNonNull(data.entries); + return data.entries; + } + + public RefreshesImpl(ServerContext context, Ample.DataLevel dataLevel) { + this.context = context; + this.dataLevel = dataLevel; + } + + private void mutateRootRefreshes(Consumer> mutator) { + String zpath = context.getZooKeeperRoot() + ZROOT_TABLET_REFRESHES; + try { + context.getZooReaderWriter().mutateExisting(zpath, currVal -> { + String currJson = new String(currVal, UTF_8); + log.debug("Root refreshes before change : {}", currJson); + Map entries = fromJson(currJson); + mutator.accept(entries); + String newJson = toJson(entries); + log.debug("Root refreshes after change : {}", newJson); + if (newJson.length() > 262_144) { + log.warn("Root refreshes stored in ZK at {} are getting large ({} bytes)." + + " Large nodes may cause problems for Zookeeper!", zpath, newJson.length()); + } + return newJson.getBytes(UTF_8); + }); + } catch (Exception e) { + throw new IllegalStateException(e); + } + } + + private Text createRow(RefreshEntry entry) { + return new Text(RefreshSection.getRowPrefix() + entry.getEcid().canonical()); + } + + private Mutation createAddMutation(RefreshEntry entry) { + Mutation m = new Mutation(createRow(entry)); + + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DataOutputStream dos = new DataOutputStream(baos)) { + dos.writeInt(1); + entry.getExtent().writeTo(dos); + dos.writeUTF(entry.getTserver().getHostPortSession()); + dos.close(); + m.at().family("").qualifier("").put(baos.toByteArray()); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + + return m; + } + + private Mutation createDeleteMutation(RefreshEntry entry) { + Mutation m = new Mutation(createRow(entry)); + m.putDelete("", ""); + return m; + } + + private void requireRootTablets(Collection entries) { + if (!entries.stream().allMatch(e -> e.getExtent().isRootTablet())) { + var nonRootTablets = entries.stream().map(RefreshEntry::getExtent) + .filter(e -> !e.isRootTablet()).collect(Collectors.toSet()); + throw new IllegalArgumentException("Expected only root tablet but saw " + nonRootTablets); + } + } + + private RefreshEntry decode(Map.Entry entry) { + String row = entry.getKey().getRowData().toString(); + Preconditions.checkArgument(row.startsWith(RefreshSection.getRowPrefix()), + "Row %s did not start with %s", row, RefreshSection.getRowPrefix()); + Preconditions.checkArgument(entry.getKey().getColumnFamilyData().length() == 0, + "Expected empty family but saw %s", entry.getKey().getColumnFamilyData()); + Preconditions.checkArgument(entry.getKey().getColumnQualifierData().length() == 0, + "Expected empty qualifier but saw %s", entry.getKey().getColumnQualifierData()); + + try (ByteArrayInputStream bais = new ByteArrayInputStream(entry.getValue().get()); + DataInputStream dis = new DataInputStream(bais)) { + var version = dis.readInt(); + Preconditions.checkArgument(version == CURRENT_VERSION, "Expected version %s saw %s", + CURRENT_VERSION, version); + var extent = KeyExtent.readFrom(dis); + var tserver = new TServerInstance(dis.readUTF()); + return new RefreshEntry( + ExternalCompactionId.of(row.substring(RefreshSection.getRowPrefix().length())), extent, + tserver); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + @Override + public void add(Collection entries) { + Objects.requireNonNull(entries); + if (dataLevel == Ample.DataLevel.ROOT) { + // expect all of these to be the root tablet, verifying because its not stored + requireRootTablets(entries); + Consumer> mutator = map -> entries.forEach(refreshEntry -> map + .put(refreshEntry.getEcid().canonical(), refreshEntry.getTserver().getHostPortSession())); + mutateRootRefreshes(mutator); + } else { + try (BatchWriter writer = context.createBatchWriter(dataLevel.metaTable())) { + for (RefreshEntry entry : entries) { + writer.addMutation(createAddMutation(entry)); + } + } catch (MutationsRejectedException | TableNotFoundException e) { + throw new IllegalStateException(e); + } + } + } + + @Override + public void delete(Collection entries) { + Objects.requireNonNull(entries); + if (dataLevel == Ample.DataLevel.ROOT) { + // expect all of these to be the root tablet, verifying because its not stored + requireRootTablets(entries); + Consumer> mutator = + map -> entries.forEach(refreshEntry -> map.remove(refreshEntry.getEcid().canonical())); + mutateRootRefreshes(mutator); + } else { + try (BatchWriter writer = context.createBatchWriter(dataLevel.metaTable())) { + for (RefreshEntry entry : entries) { + writer.addMutation(createDeleteMutation(entry)); + } + } catch (MutationsRejectedException | TableNotFoundException e) { + throw new IllegalStateException(e); + } + } + } + + @Override + public Stream stream() { + if (dataLevel == Ample.DataLevel.ROOT) { + var zooReader = context.getZooReader(); + byte[] jsonBytes; + try { + jsonBytes = + zooReader.getData(context.getZooKeeperRoot() + RootTable.ZROOT_TABLET_REFRESHES); + return fromJson(new String(jsonBytes, UTF_8)).entrySet().stream() + .map(e -> new RefreshEntry(ExternalCompactionId.of(e.getKey()), RootTable.EXTENT, + new TServerInstance(e.getValue()))); + } catch (KeeperException | InterruptedException e) { + throw new IllegalStateException(e); + } + } else { + Range range = RefreshSection.getRange(); + + Scanner scanner; + try { + scanner = context.createScanner(dataLevel.metaTable(), Authorizations.EMPTY); + } catch (TableNotFoundException e) { + throw new IllegalStateException(e); + } + scanner.setRange(range); + return scanner.stream().onClose(scanner::close).map(this::decode); + } + } +} diff --git a/server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java b/server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java index 25f4ff12ebc..0421d5bd007 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java +++ b/server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java @@ -97,6 +97,7 @@ public ConditionalTabletsMutator conditionallyMutateTablets() { private void mutateRootGcCandidates(Consumer mutator) { String zpath = context.getZooKeeperRoot() + ZROOT_TABLET_GC_CANDIDATES; try { + // TODO calling create seems unnecessary and is possibly racy and inefficient context.getZooReaderWriter().mutateOrCreate(zpath, new byte[0], currVal -> { String currJson = new String(currVal, UTF_8); RootGcCandidates rgcc = new RootGcCandidates(currJson); @@ -362,4 +363,9 @@ public void deleteScanServerFileReferences(Collection r } } + @Override + public Refreshes refreshes(DataLevel dataLevel) { + return new RefreshesImpl(context, dataLevel); + } + } diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java index 2c0bc240009..fc576ae4e37 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java @@ -27,18 +27,27 @@ import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOCATION; import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.OPID; import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.PREV_ROW; +import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.SCANS; import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.SELECTED; import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -70,6 +79,7 @@ import org.apache.accumulo.core.metadata.StoredTabletFile; import org.apache.accumulo.core.metadata.TServerInstance; import org.apache.accumulo.core.metadata.schema.Ample; +import org.apache.accumulo.core.metadata.schema.Ample.Refreshes.RefreshEntry; import org.apache.accumulo.core.metadata.schema.DataFileValue; import org.apache.accumulo.core.metadata.schema.ExternalCompactionId; import org.apache.accumulo.core.metadata.schema.ExternalCompactionMetadata; @@ -85,8 +95,8 @@ import org.apache.accumulo.core.tabletserver.thrift.TCompactionKind; import org.apache.accumulo.core.tabletserver.thrift.TCompactionStats; import org.apache.accumulo.core.tabletserver.thrift.TExternalCompactionJob; +import org.apache.accumulo.core.tabletserver.thrift.TTabletRefresh; import org.apache.accumulo.core.tabletserver.thrift.TabletServerClientService; -import org.apache.accumulo.core.trace.TraceUtil; import org.apache.accumulo.core.util.Pair; import org.apache.accumulo.core.util.Retry; import org.apache.accumulo.core.util.UtilWaitThread; @@ -95,6 +105,7 @@ import org.apache.accumulo.core.util.compaction.RunningCompaction; import org.apache.accumulo.core.util.threads.ThreadPools; import org.apache.accumulo.manager.compaction.queue.CompactionJobQueues; +import org.apache.accumulo.manager.tableOps.bulkVer2.TabletRefresher; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.compaction.CompactionConfigStorage; import org.apache.accumulo.server.compaction.CompactionPluginUtils; @@ -111,8 +122,10 @@ import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; import com.google.common.base.Preconditions; +import com.google.common.collect.Iterators; import com.google.common.collect.Sets; import com.google.common.net.HostAndPort; +import com.google.common.util.concurrent.MoreExecutors; public class CompactionCoordinator implements CompactionCoordinatorService.Iface, Runnable { @@ -129,6 +142,13 @@ public class CompactionCoordinator implements CompactionCoordinatorService.Iface protected static final Map RUNNING_CACHE = new ConcurrentHashMap<>(); + /* + * When the manager starts up any refreshes that were in progress when the last manager process + * died must be completed before new refresh entries are written. This map of countdown latches + * helps achieve that goal. + */ + private final Map refreshLatches; + private static final Cache COMPLETED = Caffeine.newBuilder().maximumSize(200).expireAfterWrite(10, TimeUnit.MINUTES).build(); @@ -152,8 +172,14 @@ public CompactionCoordinator(ServerContext ctx, LiveTServerSet tservers, this.schedExecutor = this.ctx.getScheduledExecutor(); this.security = security; this.jobQueues = jobQueues; - startCompactionCleaner(schedExecutor); - startRunningCleaner(schedExecutor); + + var refreshLatches = new EnumMap(Ample.DataLevel.class); + refreshLatches.put(Ample.DataLevel.ROOT, new CountDownLatch(1)); + refreshLatches.put(Ample.DataLevel.METADATA, new CountDownLatch(1)); + refreshLatches.put(Ample.DataLevel.USER, new CountDownLatch(1)); + this.refreshLatches = Collections.unmodifiableMap(refreshLatches); + + // At this point the manager does not have its lock so no actions should be taken yet } public void shutdown() { @@ -172,9 +198,62 @@ protected void startRunningCleaner(ScheduledThreadPoolExecutor schedExecutor) { ThreadPools.watchNonCriticalScheduledTask(future); } + private void processRefreshes(Ample.DataLevel dataLevel) { + try (var refreshStream = ctx.getAmple().refreshes(dataLevel).stream()) { + // process batches of refresh entries to avoid reading all into memory at once + Iterators.partition(refreshStream.iterator(), 10000).forEachRemaining(refreshEntries -> { + LOG.info("Processing {} tablet refreshes for {}", refreshEntries.size(), dataLevel); + + var extents = + refreshEntries.stream().map(RefreshEntry::getExtent).collect(Collectors.toList()); + var tabletsMeta = new HashMap(); + try (var tablets = ctx.getAmple().readTablets().forTablets(extents, Optional.empty()) + .fetch(PREV_ROW, LOCATION, SCANS).build()) { + tablets.stream().forEach(tm -> tabletsMeta.put(tm.getExtent(), tm)); + } + + var tserverRefreshes = new HashMap>(); + + refreshEntries.forEach(refreshEntry -> { + var tm = tabletsMeta.get(refreshEntry.getExtent()); + + // only need to refresh if the tablet is still on the same tserver instance + if (tm != null && tm.getLocation() != null + && tm.getLocation().getServerInstance().equals(refreshEntry.getTserver())) { + KeyExtent extent = tm.getExtent(); + Collection scanfiles = tm.getScans(); + var ttr = TabletRefresher.createThriftRefresh(extent, scanfiles); + tserverRefreshes.computeIfAbsent(tm.getLocation(), k -> new ArrayList<>()).add(ttr); + } + }); + + String logId = "Coordinator:" + dataLevel; + ThreadPoolExecutor threadPool = + ctx.threadPools().createFixedThreadPool(10, "Tablet refresh " + logId, false); + try { + TabletRefresher.refreshTablets(threadPool, logId, ctx, tserverSet::getCurrentServers, + tserverRefreshes); + } finally { + threadPool.shutdownNow(); + } + + ctx.getAmple().refreshes(dataLevel).delete(refreshEntries); + }); + } + // allow new refreshes to be written now that all preexisting ones are processed + refreshLatches.get(dataLevel).countDown(); + } + @Override public void run() { + processRefreshes(Ample.DataLevel.ROOT); + processRefreshes(Ample.DataLevel.METADATA); + processRefreshes(Ample.DataLevel.USER); + + startCompactionCleaner(schedExecutor); + startRunningCleaner(schedExecutor); + // On a re-start of the coordinator it's possible that external compactions are in-progress. // Attempt to get the running compactions on the compactors and then resolve which tserver // the external compaction came from to re-populate the RUNNING collection. @@ -491,6 +570,58 @@ TExternalCompactionJob createThriftJob(String externalCompactionId, ecm.getCompactionId() == null ? 0 : ecm.getCompactionId(), overrides); } + class RefreshWriter { + + private final ExternalCompactionId ecid; + private final KeyExtent extent; + + private RefreshEntry writtenEntry; + + RefreshWriter(ExternalCompactionId ecid, KeyExtent extent) { + this.ecid = ecid; + this.extent = extent; + + var dataLevel = Ample.DataLevel.of(extent.tableId()); + try { + // Wait for any refresh entries from the previous manager process to be processed before + // writing new ones. + refreshLatches.get(dataLevel).await(); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + + public void writeRefresh(TabletMetadata.Location location) { + Objects.requireNonNull(location); + + if (writtenEntry != null) { + if (location.getServerInstance().equals(writtenEntry.getTserver())) { + // the location was already written so nothing to do + return; + } else { + deleteRefresh(); + } + } + + var entry = new RefreshEntry(ecid, extent, location.getServerInstance()); + + ctx.getAmple().refreshes(Ample.DataLevel.of(extent.tableId())).add(List.of(entry)); + + LOG.debug("wrote refresh entry for {}", ecid); + + writtenEntry = entry; + } + + public void deleteRefresh() { + if (writtenEntry != null) { + ctx.getAmple().refreshes(Ample.DataLevel.of(extent.tableId())) + .delete(List.of(writtenEntry)); + LOG.debug("deleted refresh entry for {}", ecid); + writtenEntry = null; + } + } + } + private Optional getCompactionConfig(CompactionJobQueues.MetaJob metaJob) { Optional compactionConfig = Optional.empty(); @@ -510,7 +641,51 @@ private Optional getCompactionConfig(CompactionJobQueues.MetaJ } /** - * Compactor calls compactionCompleted passing in the CompactionStats + * Compactors calls this method when they have finished a compaction. This method does the + * following. + * + *
    + *
  1. Reads the tablets metadata and determines if the compaction can commit. Its possible that + * things changed while the compaction was running and it can no longer commit.
    2. + *
  2. If the compaction can commit then a ~refresh entry may be written to the metadata table. + * This is done before attempting to commit to cover the case of process failure after commit. If + * the manager dies after commit then when it restarts it will see the ~refresh entry and refresh + * that tablet. The ~refresh entry is only written when its a system compaction on a tablet with a + * location.
    3. + *
  3. Commit the compaction using a conditional mutation. If the tablets files or location + * changed since reading the tablets metadata, then conditional mutation will fail. When this + * happens it will reread the metadata and go back to step 1 conceptually. When committing a + * compaction the compacted files are removed and scan entries are added to the tablet in case the + * files are in use, this prevents GC from deleting the files between updating tablet metadata and + * refreshing the tablet. The scan entries are only added when a tablet has a location.
    4. + *
  4. After successful commit a refresh request is sent to the tablet if it has a location. This + * will cause the tablet to start using the newly compacted files for future scans. Also the + * tablet can delete the scan entries if there are no active scans using them.
    5. + *
  5. If a ~refresh entry was written, delete it since the refresh was successful.
    6. + *
+ * + *

+ * User compactions will be refreshed as part of the fate operation. The user compaction fate + * operation will see the compaction was committed after this code updates the tablet metadata, + * however if it were to rely on this code to do the refresh it would not be able to know when the + * refresh was actually done. Therefore, user compactions will refresh as part of the fate + * operation so that it's known to be done before the fate operation returns. Since the fate + * operation will do it, there is no need to do it here for user compactions. + *

+ * + *

+ * The ~refresh entries serve a similar purpose to FATE operations, it ensures that code executes + * even when a process dies. FATE was intentionally not used for compaction commit because FATE + * stores its data in zookeeper. The refresh entry is stored in the metadata table, which is much + * more scalable than zookeeper. The number of system compactions of small files could be large + * and this would be a large number of writes to zookeeper. Zookeeper scales somewhat with reads, + * but not with writes. + *

+ * + *

+ * Issue #3559 was opened to explore the possibility of making compaction commit a fate operation + * which would remove the need for the ~refresh section. + *

* * @param tinfo trace info * @param credentials tcredentials object @@ -558,15 +733,21 @@ public void compactionCompleted(TInfo tinfo, TCredentials credentials, return; } + RefreshWriter refreshWriter = new RefreshWriter(ecid, extent); + try { - tabletMeta = commitCompaction(stats, ecid, tabletMeta, optionalNewFile); + tabletMeta = commitCompaction(stats, ecid, tabletMeta, optionalNewFile, refreshWriter); } catch (RuntimeException e) { LOG.warn("Failed to commit complete compaction {} {}", ecid, extent, e); compactionFailed(Map.of(ecid, extent)); } - // ELASTICITY_TODO, how will user compactions handle refresh - refreshTablet(tabletMeta); + if (ecm.getKind() != CompactionKind.USER) { + refreshTablet(tabletMeta, ecm.getJobFiles()); + } + + // if a refresh entry was written, it can be removed after the tablet was refreshed + refreshWriter.deleteRefresh(); // It's possible that RUNNING might not have an entry for this ecid in the case // of a coordinator restart when the Coordinator can't find the TServer for the @@ -594,18 +775,21 @@ private Optional renameOrDeleteFile(TCompactionStats stats } } - private void refreshTablet(TabletMetadata metadata) { + private void refreshTablet(TabletMetadata metadata, Collection scanfiles) { var location = metadata.getLocation(); if (location != null) { - TabletServerClientService.Client client = null; + KeyExtent extent = metadata.getExtent(); + TTabletRefresh tTabletRefresh = TabletRefresher.createThriftRefresh(extent, scanfiles); + + // there is a single tserver and single tablet, do not need a thread pool. The direct executor + // will run everything in the current thread + ExecutorService executorService = MoreExecutors.newDirectExecutorService(); try { - client = getTabletServerConnection(location.getServerInstance()); - client.refreshTablets(TraceUtil.traceInfo(), ctx.rpcCreds(), - List.of(metadata.getExtent().toThrift())); - } catch (TException e) { - throw new RuntimeException(e); + TabletRefresher.refreshTablets(executorService, + "compaction:" + metadata.getExtent().toString(), ctx, tserverSet::getCurrentServers, + Map.of(metadata.getLocation(), List.of(tTabletRefresh))); } finally { - returnTServerClient(client); + executorService.shutdownNow(); } } } @@ -665,7 +849,8 @@ private boolean canCommitCompaction(ExternalCompactionId ecid, TabletMetadata ta } private TabletMetadata commitCompaction(TCompactionStats stats, ExternalCompactionId ecid, - TabletMetadata tablet, Optional newDatafile) { + TabletMetadata tablet, Optional newDatafile, + RefreshWriter refreshWriter) { KeyExtent extent = tablet.getExtent(); @@ -676,9 +861,18 @@ private TabletMetadata commitCompaction(TCompactionStats stats, ExternalCompacti while (canCommitCompaction(ecid, tablet)) { ExternalCompactionMetadata ecm = tablet.getExternalCompactions().get(ecid); + if (tablet.getLocation() != null + && tablet.getExternalCompactions().get(ecid).getKind() != CompactionKind.USER) { + // Write the refresh entry before attempting to update tablet metadata, this ensures that + // refresh will happen even if this process dies. In the case where this process does not + // die refresh will happen after commit. User compactions will make refresh calls in their + // fate operation, so it does not need to be done here. + refreshWriter.writeRefresh(tablet.getLocation()); + } + try (var tabletsMutator = ctx.getAmple().conditionallyMutateTablets()) { var tabletMutator = tabletsMutator.mutateTablet(extent).requireAbsentOperation() - .requireCompaction(ecid).requireSame(tablet, PREV_ROW, FILES); + .requireCompaction(ecid).requireSame(tablet, PREV_ROW, FILES, LOCATION); if (ecm.getKind() == CompactionKind.USER || ecm.getKind() == CompactionKind.SELECTOR) { tabletMutator.requireSame(tablet, SELECTED, COMPACT_ID); @@ -764,6 +958,11 @@ private void updateTabletForCompaction(TCompactionStats stats, ExternalCompactio } } + if (tablet.getLocation() != null) { + // add scan entries to prevent GC in case the hosted tablet is currently using the files for + // scan + ecm.getJobFiles().forEach(tabletMutator::putScan); + } ecm.getJobFiles().forEach(tabletMutator::deleteFile); tabletMutator.deleteExternalCompaction(ecid); diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/RefreshTablets.java b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/RefreshTablets.java index edfa3664f81..de7c2961885 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/RefreshTablets.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/RefreshTablets.java @@ -18,38 +18,9 @@ */ package org.apache.accumulo.manager.tableOps.bulkVer2; -import static java.util.concurrent.TimeUnit.MILLISECONDS; -import static java.util.concurrent.TimeUnit.MINUTES; -import static java.util.concurrent.TimeUnit.SECONDS; -import static java.util.stream.Collectors.groupingBy; -import static java.util.stream.Collectors.mapping; -import static java.util.stream.Collectors.toList; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.ThreadPoolExecutor; - -import org.apache.accumulo.core.conf.Property; -import org.apache.accumulo.core.dataImpl.KeyExtent; -import org.apache.accumulo.core.fate.FateTxId; import org.apache.accumulo.core.fate.Repo; -import org.apache.accumulo.core.metadata.TServerInstance; -import org.apache.accumulo.core.metadata.schema.TabletMetadata; -import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType; -import org.apache.accumulo.core.metadata.schema.TabletMetadata.Location; -import org.apache.accumulo.core.rpc.ThriftUtil; -import org.apache.accumulo.core.rpc.clients.ThriftClientTypes; -import org.apache.accumulo.core.tabletserver.thrift.TabletServerClientService; -import org.apache.accumulo.core.trace.TraceUtil; -import org.apache.accumulo.core.util.Retry; import org.apache.accumulo.manager.Manager; import org.apache.accumulo.manager.tableOps.ManagerRepo; -import org.apache.accumulo.server.manager.LiveTServerSet; -import org.apache.thrift.TException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -80,135 +51,10 @@ public long isReady(long tid, Manager manager) throws Exception { @Override public Repo call(long tid, Manager manager) throws Exception { - Map> refreshesNeeded; - - try (var tablets = manager.getContext().getAmple().readTablets().forTable(bulkInfo.tableId) - .overlapping(bulkInfo.firstSplit, bulkInfo.lastSplit).checkConsistency() - .fetch(ColumnType.LOADED, ColumnType.LOCATION, ColumnType.PREV_ROW).build()) { - - // Find all tablets that have a location and load markers for this bulk load operation and - // therefore need to refresh their metadata. There may be some tablets in the map that were - // hosted after the bulk files were set and that is ok, it just results in an unneeded refresh - // request. There may also be tablets that had a location when the files were set but do not - // have a location now, that is ok the next time that tablet loads somewhere it will see the - // files. - refreshesNeeded = tablets.stream() - .filter(tabletMetadata -> tabletMetadata.getLocation() != null - && tabletMetadata.getLoaded().containsValue(tid)) - .collect(groupingBy(TabletMetadata::getLocation, - mapping(TabletMetadata::getExtent, toList()))); - } + TabletRefresher.refresh(manager.getContext(), manager::onlineTabletServers, tid, + bulkInfo.tableId, bulkInfo.firstSplit, bulkInfo.lastSplit, + tabletMetadata -> tabletMetadata.getLoaded().containsValue(tid)); - if (!refreshesNeeded.isEmpty()) { - // block until all tablets are refreshed - refreshTablets(tid, manager, refreshesNeeded); - } return new CleanUpBulkImport(bulkInfo); } - - private void refreshTablets(long tid, Manager manager, - Map> refreshesNeeded) - throws InterruptedException, ExecutionException { - - // ELASTICITY_TODO should this thread pool be configurable? - ThreadPoolExecutor threadPool = manager.getContext().threadPools().createFixedThreadPool(5, - "Bulk load refresh" + FateTxId.formatTid(tid), false); - - try { - Retry retry = Retry.builder().infiniteRetries().retryAfter(100, MILLISECONDS) - .incrementBy(100, MILLISECONDS).maxWait(1, SECONDS).backOffFactor(1.5) - .logInterval(3, MINUTES).createRetry(); - - while (!refreshesNeeded.isEmpty()) { - - Map>> futures = new HashMap<>(); - - for (Map.Entry> entry : refreshesNeeded.entrySet()) { - - // Ask tablet server to reload the metadata for these tablets. The tablet server returns - // the list of extents it was hosting but was unable to refresh (the tablets could be in - // the process of loading). If it is not currently hosting the tablet it treats that as - // refreshed and does not return anything for it. - Future> future = threadPool - .submit(() -> sendSyncRefreshRequest(manager, tid, entry.getKey(), entry.getValue())); - - futures.put(entry.getKey(), future); - } - - for (Map.Entry>> entry : futures.entrySet()) { - Location location = entry.getKey(); - Future> future = entry.getValue(); - - List nonRefreshedExtents = future.get(); - if (nonRefreshedExtents.isEmpty()) { - // tablet server was able to refresh everything, so remove that location - refreshesNeeded.remove(location); - } else { - // tablet server could not refresh some tablets, try them again later. - refreshesNeeded.put(location, nonRefreshedExtents); - } - } - - // look for any tservers that have died since we read the metadata table and remove them - if (!refreshesNeeded.isEmpty()) { - LiveTServerSet lts = - new LiveTServerSet(manager.getContext(), (current, deleted, added) -> {}); - Set liveTservers = lts.getCurrentServers(); - - refreshesNeeded.keySet() - .removeIf(location -> !liveTservers.contains(location.getServerInstance())); - } - - if (!refreshesNeeded.isEmpty()) { - retry.waitForNextAttempt(log, FateTxId.formatTid(tid) + " waiting for " - + refreshesNeeded.size() + " tservers to refresh their tablets metadata"); - } - } - } finally { - threadPool.shutdownNow(); - } - } - - private List sendSyncRefreshRequest(Manager manager, long tid, Location location, - List extents) { - TabletServerClientService.Client client = null; - try { - log.trace("{} sending refresh request to {} for {} extents", FateTxId.formatTid(tid), - location, extents.size()); - var timeInMillis = manager.getConfiguration().getTimeInMillis(Property.MANAGER_BULK_TIMEOUT); - client = ThriftUtil.getClient(ThriftClientTypes.TABLET_SERVER, location.getHostAndPort(), - manager.getContext(), timeInMillis); - - var unrefreshed = - client.refreshTablets(TraceUtil.traceInfo(), manager.getContext().rpcCreds(), - extents.stream().map(KeyExtent::toThrift).collect(toList())); - - log.trace("{} refresh request to {} returned {} unrefreshed extents", FateTxId.formatTid(tid), - location, unrefreshed.size()); - - var unrefreshedConverted = unrefreshed.stream().map(KeyExtent::fromThrift).collect(toList()); - - if (log.isDebugEnabled() && !unrefreshedConverted.isEmpty()) { - // this situation should be fairly uncommon, but could hold up bulk import. so lets log some - // info about it in case it gets stuck on a particular tablet - var fateStr = FateTxId.formatTid(tid); - for (var extent : unrefreshedConverted) { - log.debug("{} tserver {} was unable to refresh tablet {}", fateStr, location, extent); - } - } - - return unrefreshedConverted; - } catch (TException ex) { - var fmtTid = FateTxId.formatTid(tid); - log.debug("rpc failed server: " + location + ", " + fmtTid + " " + ex.getMessage(), ex); - - // ELASTICITY_TODO are there any other exceptions we should catch in this method and check if - // the tserver is till alive? - - // something went wrong w/ RPC return all extents as unrefreshed - return extents; - } finally { - ThriftUtil.returnClient(client, manager.getContext()); - } - } } diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/TabletRefresher.java b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/TabletRefresher.java new file mode 100644 index 00000000000..3555c82e358 --- /dev/null +++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/TabletRefresher.java @@ -0,0 +1,212 @@ +/* + * 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.manager.tableOps.bulkVer2; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.MINUTES; +import static java.util.concurrent.TimeUnit.SECONDS; +import static java.util.stream.Collectors.groupingBy; +import static java.util.stream.Collectors.mapping; +import static java.util.stream.Collectors.toList; + +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.function.Predicate; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import org.apache.accumulo.core.conf.Property; +import org.apache.accumulo.core.data.TableId; +import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent; +import org.apache.accumulo.core.fate.FateTxId; +import org.apache.accumulo.core.metadata.StoredTabletFile; +import org.apache.accumulo.core.metadata.TServerInstance; +import org.apache.accumulo.core.metadata.schema.TabletMetadata; +import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType; +import org.apache.accumulo.core.rpc.ThriftUtil; +import org.apache.accumulo.core.rpc.clients.ThriftClientTypes; +import org.apache.accumulo.core.tabletserver.thrift.TTabletRefresh; +import org.apache.accumulo.core.tabletserver.thrift.TabletServerClientService; +import org.apache.accumulo.core.trace.TraceUtil; +import org.apache.accumulo.core.util.Retry; +import org.apache.accumulo.server.ServerContext; +import org.apache.thrift.TException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.collect.Iterators; + +public class TabletRefresher { + + private static final Logger log = LoggerFactory.getLogger(TabletRefresher.class); + + public static void refresh(ServerContext context, + Supplier> onlineTserversSupplier, long fateTxid, TableId tableId, + byte[] startRow, byte[] endRow, Predicate needsRefresh) { + + // ELASTICITY_TODO should this thread pool be configurable? + ThreadPoolExecutor threadPool = context.threadPools().createFixedThreadPool(10, + "Tablet refresh " + FateTxId.formatTid(fateTxid), false); + + try (var tablets = context.getAmple().readTablets().forTable(tableId) + .overlapping(startRow, endRow).checkConsistency() + .fetch(ColumnType.LOADED, ColumnType.LOCATION, ColumnType.PREV_ROW, ColumnType.SCANS) + .build()) { + + // Find all tablets that need to refresh their metadata. There may be some tablets that were + // hosted after the tablet files were updated, it just results in an unneeded refresh + // request. There may also be tablets that had a location when the files were set but do not + // have a location now, that is ok the next time that tablet loads somewhere it will see the + // files. + + var tabletIterator = + tablets.stream().filter(tabletMetadata -> tabletMetadata.getLocation() != null) + .filter(needsRefresh).iterator(); + + // avoid reading all tablets into memory and instead process batches of 1000 tablets at a time + Iterators.partition(tabletIterator, 1000).forEachRemaining(batch -> { + var refreshesNeeded = batch.stream() + .collect(groupingBy(TabletMetadata::getLocation, + mapping(tabletMetadata -> createThriftRefresh(tabletMetadata.getExtent(), + tabletMetadata.getScans()), toList()))); + + refreshTablets(threadPool, FateTxId.formatTid(fateTxid), context, onlineTserversSupplier, + refreshesNeeded); + }); + + } finally { + threadPool.shutdownNow(); + } + + } + + public static void refreshTablets(ExecutorService threadPool, String logId, ServerContext context, + Supplier> onlineTserversSupplier, + Map> refreshesNeeded) { + + Retry retry = Retry.builder().infiniteRetries().retryAfter(100, MILLISECONDS) + .incrementBy(100, MILLISECONDS).maxWait(1, SECONDS).backOffFactor(1.5) + .logInterval(3, MINUTES).createRetry(); + + while (!refreshesNeeded.isEmpty()) { + + Map>> futures = new HashMap<>(); + + for (Map.Entry> entry : refreshesNeeded + .entrySet()) { + + // Ask tablet server to reload the metadata for these tablets. The tablet server returns + // the list of extents it was hosting but was unable to refresh (the tablets could be in + // the process of loading). If it is not currently hosting the tablet it treats that as + // refreshed and does not return anything for it. + Future> future = threadPool + .submit(() -> sendSyncRefreshRequest(context, logId, entry.getKey(), entry.getValue())); + + futures.put(entry.getKey(), future); + } + + for (Map.Entry>> entry : futures + .entrySet()) { + TabletMetadata.Location location = entry.getKey(); + Future> future = entry.getValue(); + + List nonRefreshedExtents = null; + try { + nonRefreshedExtents = future.get(); + } catch (InterruptedException | ExecutionException e) { + throw new RuntimeException(e); + } + if (nonRefreshedExtents.isEmpty()) { + // tablet server was able to refresh everything, so remove that location + refreshesNeeded.remove(location); + } else { + // tablet server could not refresh some tablets, try them again later. + refreshesNeeded.put(location, nonRefreshedExtents); + } + } + + // look for any tservers that have died since we read the metadata table and remove them + if (!refreshesNeeded.isEmpty()) { + Set liveTservers = onlineTserversSupplier.get(); + + refreshesNeeded.keySet() + .removeIf(location -> !liveTservers.contains(location.getServerInstance())); + } + + if (!refreshesNeeded.isEmpty()) { + try { + retry.waitForNextAttempt(log, logId + " waiting for " + refreshesNeeded.size() + + " tservers to refresh their tablets metadata"); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + } + } + + private static List sendSyncRefreshRequest(ServerContext context, String logId, + TabletMetadata.Location location, List refreshes) { + TabletServerClientService.Client client = null; + try { + log.trace("{} sending refresh request to {} for {} extents", logId, location, + refreshes.size()); + var timeInMillis = context.getConfiguration().getTimeInMillis(Property.MANAGER_BULK_TIMEOUT); + client = ThriftUtil.getClient(ThriftClientTypes.TABLET_SERVER, location.getHostAndPort(), + context, timeInMillis); + + var unrefreshed = client.refreshTablets(TraceUtil.traceInfo(), context.rpcCreds(), refreshes); + + log.trace("{} refresh request to {} returned {} unrefreshed extents", logId, location, + unrefreshed.size()); + + if (unrefreshed.isEmpty()) { + return List.of(); + } + + Map unrefreshedMap = new HashMap<>(); + refreshes.forEach(ttr -> unrefreshedMap.put(ttr.getExtent(), ttr)); + unrefreshedMap.keySet().retainAll(unrefreshed); + return List.copyOf(unrefreshedMap.values()); + } catch (TException ex) { + log.debug("rpc failed server: " + location + ", " + logId + " " + ex.getMessage(), ex); + + // ELASTICITY_TODO are there any other exceptions we should catch in this method and check if + // the tserver is till alive? + + // something went wrong w/ RPC return all extents as unrefreshed + return refreshes; + } finally { + ThriftUtil.returnClient(client, context); + } + } + + public static TTabletRefresh createThriftRefresh(KeyExtent extent, + Collection scanfiles) { + return new TTabletRefresh(extent.toThrift(), + scanfiles.stream().map(StoredTabletFile::getMetaUpdateDelete).collect(Collectors.toList())); + } +} diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/compact/CompactionDriver.java b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/compact/CompactionDriver.java index 8fc8fed77ce..51e39f96974 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/compact/CompactionDriver.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/compact/CompactionDriver.java @@ -49,7 +49,6 @@ import org.apache.accumulo.core.util.Pair; import org.apache.accumulo.manager.Manager; import org.apache.accumulo.manager.tableOps.ManagerRepo; -import org.apache.accumulo.manager.tableOps.Utils; import org.apache.accumulo.manager.tableOps.delete.PreDeleteTable; import org.apache.accumulo.server.compaction.CompactionConfigStorage; import org.apache.accumulo.server.compaction.CompactionPluginUtils; @@ -247,10 +246,7 @@ public int updateAndCheckTablets(Manager manager, long tid, long compactId) { @Override public Repo call(long tid, Manager env) throws Exception { - CompactRange.removeIterators(env, tid, tableId); - Utils.getReadLock(env, tableId, tid).unlock(); - Utils.getReadLock(env, namespaceId, tid).unlock(); - return null; + return new RefreshTablets(tableId, namespaceId, startRow, endRow); } @Override diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/compact/RefreshTablets.java b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/compact/RefreshTablets.java new file mode 100644 index 00000000000..a9b3f85e16b --- /dev/null +++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/compact/RefreshTablets.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.accumulo.manager.tableOps.compact; + +import org.apache.accumulo.core.data.NamespaceId; +import org.apache.accumulo.core.data.TableId; +import org.apache.accumulo.core.fate.Repo; +import org.apache.accumulo.manager.Manager; +import org.apache.accumulo.manager.tableOps.ManagerRepo; +import org.apache.accumulo.manager.tableOps.Utils; +import org.apache.accumulo.manager.tableOps.bulkVer2.TabletRefresher; + +public class RefreshTablets extends ManagerRepo { + + private static final long serialVersionUID = 1L; + + private final TableId tableId; + private final NamespaceId namespaceId; + private final byte[] startRow; + private final byte[] endRow; + + public RefreshTablets(TableId tableId, NamespaceId namespaceId, byte[] startRow, byte[] endRow) { + this.tableId = tableId; + this.namespaceId = namespaceId; + this.startRow = startRow; + this.endRow = endRow; + } + + @Override + public Repo call(long tid, Manager manager) throws Exception { + TabletRefresher.refresh(manager.getContext(), manager::onlineTabletServers, tid, tableId, + startRow, endRow, tabletMetadata -> true); + + CompactRange.removeIterators(manager, tid, tableId); + Utils.getReadLock(manager, tableId, tid).unlock(); + Utils.getReadLock(manager, namespaceId, tid).unlock(); + + return null; + } +} diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java index 1e80e57d6d4..533734a4a03 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java @@ -79,6 +79,7 @@ import org.apache.accumulo.core.metadata.MetadataTable; import org.apache.accumulo.core.metadata.ReferencedTabletFile; import org.apache.accumulo.core.metadata.RootTable; +import org.apache.accumulo.core.metadata.StoredTabletFile; import org.apache.accumulo.core.metadata.schema.ExternalCompactionId; import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.core.securityImpl.thrift.TCredentials; @@ -97,6 +98,7 @@ import org.apache.accumulo.core.tabletserver.thrift.NotServingTabletException; import org.apache.accumulo.core.tabletserver.thrift.TCompactionQueueSummary; import org.apache.accumulo.core.tabletserver.thrift.TExternalCompactionJob; +import org.apache.accumulo.core.tabletserver.thrift.TTabletRefresh; import org.apache.accumulo.core.tabletserver.thrift.TabletServerClientService; import org.apache.accumulo.core.tabletserver.thrift.TabletStats; import org.apache.accumulo.core.trace.TraceUtil; @@ -1372,7 +1374,7 @@ public void compactionJobFailed(TInfo tinfo, TCredentials credentials, @Override public List refreshTablets(TInfo tinfo, TCredentials credentials, - List extents) throws TException { + List refreshes) throws TException { if (!security.canPerformSystemActions(credentials)) { throw new AccumuloSecurityException(credentials.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED).asThriftException(); @@ -1382,10 +1384,10 @@ public List refreshTablets(TInfo tinfo, TCredentials credentials, // handle that more expensive case if needed. var tabletsSnapshot = server.getOnlineTablets(); - Set notFound = new HashSet<>(); + Map notFound = new HashMap<>(); - for (var tExtent : extents) { - var extent = KeyExtent.fromThrift(tExtent); + for (var tTabletRefresh : refreshes) { + var extent = KeyExtent.fromThrift(tTabletRefresh.getExtent()); var tablet = tabletsSnapshot.get(extent); if (tablet != null) { @@ -1394,9 +1396,10 @@ public List refreshTablets(TInfo tinfo, TCredentials credentials, // and multiple concurrent refresh request), so defer doing this until after removing // functionality from the tablet. No need to make the change now and have to change it // later. - tablet.refresh(); + tablet.refresh( + tTabletRefresh.getScanEntries().stream().map(StoredTabletFile::new).collect(toList())); } else { - notFound.add(extent); + notFound.put(extent, tTabletRefresh); } } @@ -1412,7 +1415,7 @@ public List refreshTablets(TInfo tinfo, TCredentials credentials, // Get the snapshot again, however this time nothing will be changing while we iterate // over the snapshot because all three locks are held. tabletsSnapshot = server.getOnlineTablets(); - for (var extent : notFound) { + for (var extent : notFound.keySet()) { // TODO investigate if its safe to ignore tablets in the unopened set because they // have not yet read any metadata if (server.unopenedTablets.contains(extent) @@ -1439,7 +1442,8 @@ public List refreshTablets(TInfo tinfo, TCredentials credentials, } for (var tablet : foundTablets) { - tablet.refresh(); + tablet.refresh(notFound.get(tablet.getExtent()).getScanEntries().stream() + .map(StoredTabletFile::new).collect(toList())); } if (log.isDebugEnabled()) { diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java index 5bb90e551b0..f7c08086b3f 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java @@ -2105,7 +2105,7 @@ public boolean isOnDemand() { return goal == TabletHostingGoal.ONDEMAND; } - public void refresh() { + public void refresh(List scanEntries) { if (isClosing() || isClosed()) { // TODO this is just a best effort could close after this check, its a race condition. // Intentionally not being handled ATM. @@ -2127,5 +2127,15 @@ public void refresh() { // TODO this could have race conditions with minor compactions. Intentionally // not being handled ATM. getDatafileManager().setFilesHack(tabletMetadata.getFilesMap()); + + if (!scanEntries.isEmpty()) { + // ELASTICITY_TODO this is a temporary hack. Should not always remove scan entries added by a + // compaction, need to check and see if they are actually in use by a scan. If in use need to + // add to a set to remove later. Also should use a conditional mutation to update, did not + // bother using conditional mutation as this is temporary. + var tabletMutator = getContext().getAmple().mutateTablet(extent); + scanEntries.forEach(tabletMutator::deleteScan); + tabletMutator.mutate(); + } } } diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/RefreshesIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/RefreshesIT.java new file mode 100644 index 00000000000..053f01d5a36 --- /dev/null +++ b/test/src/main/java/org/apache/accumulo/test/compaction/RefreshesIT.java @@ -0,0 +1,97 @@ +/* + * 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.stream.Collectors.toMap; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import org.apache.accumulo.core.data.TableId; +import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.metadata.RootTable; +import org.apache.accumulo.core.metadata.TServerInstance; +import org.apache.accumulo.core.metadata.schema.Ample; +import org.apache.accumulo.core.metadata.schema.Ample.Refreshes.RefreshEntry; +import org.apache.accumulo.core.metadata.schema.ExternalCompactionId; +import org.apache.accumulo.test.functional.ConfigurableMacBase; +import org.apache.hadoop.io.Text; +import org.junit.jupiter.api.Test; + +/** + * Tests reading and writing refesh entries. This is done as a an IT instead of a unit test because + * it would take too much code to mock ZK and the metadata table. + */ +public class RefreshesIT extends ConfigurableMacBase { + private void testRefreshes(Ample.DataLevel level, KeyExtent extent1, KeyExtent extent2) { + var refreshes = getServerContext().getAmple().refreshes(level); + + assertEquals(0, refreshes.stream().count()); + + var ecid1 = ExternalCompactionId.generate(UUID.randomUUID()); + var ecid2 = ExternalCompactionId.generate(UUID.randomUUID()); + var ecid3 = ExternalCompactionId.generate(UUID.randomUUID()); + + var tserver1 = new TServerInstance("host1:9997[abcdef123]"); + var tserver2 = new TServerInstance("host2:9997[1234567890]"); + + refreshes.add(List.of(new RefreshEntry(ecid1, extent1, tserver1))); + + assertEquals(Map.of(ecid1, extent1), + refreshes.stream().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); + assertEquals(Map.of(ecid1, tserver1), + refreshes.stream().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getTserver))); + + refreshes.add(List.of(new RefreshEntry(ecid2, extent2, tserver1), + new RefreshEntry(ecid3, extent2, tserver2))); + + assertEquals(Map.of(ecid1, extent1, ecid2, extent2, ecid3, extent2), + refreshes.stream().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); + assertEquals(Map.of(ecid1, tserver1, ecid2, tserver1, ecid3, tserver2), + refreshes.stream().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getTserver))); + + refreshes.delete(List.of(new RefreshEntry(ecid2, extent2, tserver1))); + + assertEquals(Map.of(ecid1, extent1, ecid3, extent2), + refreshes.stream().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); + assertEquals(Map.of(ecid1, tserver1, ecid3, tserver2), + refreshes.stream().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getTserver))); + + refreshes.delete(List.of(new RefreshEntry(ecid3, extent2, tserver2), + new RefreshEntry(ecid1, extent1, tserver1))); + + assertEquals(0, refreshes.stream().count()); + } + + @Test + public void testRefreshStorage() { + var extent1 = new KeyExtent(TableId.of("1"), null, null); + var extent2 = new KeyExtent(TableId.of("2"), new Text("m"), new Text("c")); + + // the root level only expects the root tablet extent + assertThrows(IllegalArgumentException.class, + () -> testRefreshes(Ample.DataLevel.ROOT, extent1, extent2)); + testRefreshes(Ample.DataLevel.ROOT, RootTable.EXTENT, RootTable.EXTENT); + testRefreshes(Ample.DataLevel.METADATA, extent1, extent2); + testRefreshes(Ample.DataLevel.USER, extent1, extent2); + } +} diff --git a/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java b/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java index c9992b3ccd9..2ecff2e96e5 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java @@ -42,6 +42,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; import org.apache.accumulo.core.client.Accumulo; import org.apache.accumulo.core.client.AccumuloClient; @@ -57,11 +58,14 @@ import org.apache.accumulo.core.client.admin.compaction.CompactableFile; import org.apache.accumulo.core.client.admin.compaction.CompactionSelector; import org.apache.accumulo.core.client.admin.compaction.CompressionConfigurer; +import org.apache.accumulo.core.clientImpl.ClientContext; import org.apache.accumulo.core.clientImpl.TableOperationsImpl; import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.core.data.Key; import org.apache.accumulo.core.data.Mutation; +import org.apache.accumulo.core.data.TableId; import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.dataImpl.KeyExtent; import org.apache.accumulo.core.iterators.DevNull; import org.apache.accumulo.core.iterators.Filter; import org.apache.accumulo.core.iterators.IteratorEnvironment; @@ -730,6 +734,82 @@ public void testMetadataCompactions() throws Exception { } } + @Test + public void testSystemCompactionsRefresh() throws Exception { + // This test ensures that after a system compaction occurs that a tablet will refresh its files. + + String tableName = getUniqueNames(1)[0]; + try (final AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) { + + // configure tablet compaction iterator that filters out data not divisible by 7 and cause + // table to compact to one file + + var ntc = new NewTableConfiguration(); + IteratorSetting iterSetting = new IteratorSetting(100, TestFilter.class); + iterSetting.addOption("modulus", 7 + ""); + ntc.attachIterator(iterSetting, EnumSet.of(IteratorScope.majc)); + ntc.setProperties(Map.of(Property.TABLE_MAJC_RATIO.getKey(), "20")); + + client.tableOperations().create(tableName, ntc); + + Set expectedData = new HashSet<>(); + + // Insert MAX_DATA rows + try (BatchWriter bw = client.createBatchWriter(tableName)) { + for (int i = 0; i < MAX_DATA; i++) { + Mutation m = new Mutation(String.format("r:%04d", i)); + m.put("", "", "" + i); + bw.addMutation(m); + + if (i % 75 == 0) { + // create many files as this will cause a system compaction + bw.flush(); + client.tableOperations().flush(tableName, null, null, true); + } + + if (i % 7 == 0) { + expectedData.add(i); + } + } + } + + client.tableOperations().flush(tableName, null, null, true); + + // there should be no system compactions yet and no data should be filtered, so should see all + // data that was written + try (Scanner scanner = client.createScanner(tableName)) { + assertEquals(MAX_DATA, scanner.stream().count()); + } + + // set the compaction ratio 1 which should cause system compactions to filter data and refresh + // the tablets files + client.tableOperations().setProperty(tableName, Property.TABLE_MAJC_RATIO.getKey(), "1"); + + var tableId = TableId.of(client.tableOperations().tableIdMap().get(tableName)); + var extent = new KeyExtent(tableId, null, null); + + // wait for the compactions to filter data and refresh that tablets files + Wait.waitFor(() -> { + var tabletMeta = ((ClientContext) client).getAmple().readTablet(extent); + var files = tabletMeta.getFiles(); + log.debug("Current files {}", + files.stream().map(StoredTabletFile::getFileName).collect(Collectors.toList())); + + if (files.size() == 1) { + // Once only one file exists the tablet may still have not gotten the refresh message + // because its sent after the metadata update. After the tablet is down to one file should + // eventually see the tablet refresh its files. + try (Scanner scanner = client.createScanner(tableName)) { + var acutalData = scanner.stream().map(e -> Integer.parseInt(e.getValue().toString())) + .collect(Collectors.toSet()); + return acutalData.equals(expectedData); + } + } + return false; + }); + } + } + private int countFiles(AccumuloClient c) throws Exception { try (Scanner s = c.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) { s.fetchColumnFamily(new Text(TabletColumnFamily.NAME)); @@ -768,5 +848,4 @@ private void writeFlush(AccumuloClient client, String tablename, String row) thr } client.tableOperations().flush(tablename, null, null, true); } - } diff --git a/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java b/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java index 5d2be5df56b..e5fedb43721 100644 --- a/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java +++ b/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java @@ -71,6 +71,7 @@ import org.apache.accumulo.core.tabletserver.thrift.ActiveCompaction; import org.apache.accumulo.core.tabletserver.thrift.TCompactionQueueSummary; import org.apache.accumulo.core.tabletserver.thrift.TExternalCompactionJob; +import org.apache.accumulo.core.tabletserver.thrift.TTabletRefresh; import org.apache.accumulo.core.tabletserver.thrift.TabletServerClientService; import org.apache.accumulo.core.tabletserver.thrift.TabletStats; import org.apache.accumulo.core.util.threads.ThreadPools; @@ -298,7 +299,7 @@ public void compactionJobFailed(TInfo tinfo, TCredentials credentials, @Override public List refreshTablets(TInfo tinfo, TCredentials credentials, - List extents) throws TException { + List refreshes) throws TException { return List.of(); } }