From 3bf5e73b004a57478822e81e212a22166c0c691e Mon Sep 17 00:00:00 2001 From: Keith Turner Date: Fri, 30 Jun 2023 20:30:12 -0400 Subject: [PATCH 1/4] Reliably refresh tablets after updating tablet files for a compaction. This commit modifies the compaction coordinator to reliably refresh hosted tablets in memory metadata after a compaction. The RPC request to a hosted tablet asking it to refresh happens after updating a tablets metadata for a compaction. The changes in the PR ensure that if the manager process dies after making the metadata update but before it makes the RPC that this will happen when the manager restarts. The implementation for this relies on a new ~refresh sections in the metadata table. System compactions on hosted tablets will write a ~refresh entry before trying to update a tablets metadata. This entry allows the compaction commit process to be fault tolerant. Javadoc was added to CompactionCoordinator.compactionCompleted() describing the overall process of committing a compaction. This commit also refactored the refresh RPC code in bulk import so that it could also be used by compactions. This commit also made the tablet metadata update for a compaction add scan entries for removed files when a tablet is hosted. This was done to prevent the Accumulo GC from removing compacted files that may be in use by an immediate scan. fixes #3462 --- .../accumulo/core/metadata/RootTable.java | 4 + .../accumulo/core/metadata/schema/Ample.java | 43 ++ .../core/metadata/schema/MetadataSchema.java | 13 + .../tabletserver/thrift/TTabletRefresh.java | 566 ++++++++++++++++++ .../thrift/TabletServerClientService.java | 476 +++++++-------- core/src/main/thrift/tabletserver.thrift | 7 +- .../server/init/ZooKeeperInitializer.java | 3 + .../server/metadata/RefreshesImpl.java | 237 ++++++++ .../server/metadata/ServerAmpleImpl.java | 6 + .../coordinator/CompactionCoordinator.java | 233 ++++++- .../tableOps/bulkVer2/RefreshTablets.java | 160 +---- .../tableOps/bulkVer2/TabletRefresher.java | 212 +++++++ .../tableOps/compact/CompactionDriver.java | 6 +- .../tableOps/compact/RefreshTablets.java | 57 ++ .../accumulo/tserver/TabletClientHandler.java | 20 +- .../accumulo/tserver/tablet/Tablet.java | 12 +- .../accumulo/test/compaction/RefreshesIT.java | 97 +++ .../test/functional/CompactionIT.java | 81 ++- .../test/performance/NullTserver.java | 3 +- 19 files changed, 1807 insertions(+), 429 deletions(-) create mode 100644 core/src/main/thrift-gen-java/org/apache/accumulo/core/tabletserver/thrift/TTabletRefresh.java create mode 100644 server/base/src/main/java/org/apache/accumulo/server/metadata/RefreshesImpl.java create mode 100644 server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/TabletRefresher.java create mode 100644 server/manager/src/main/java/org/apache/accumulo/manager/tableOps/compact/RefreshTablets.java create mode 100644 test/src/main/java/org/apache/accumulo/test/compaction/RefreshesIT.java 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..6398b44283e 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 gc candidates 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..09f2e40aed6 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 list(); + } + + /** + * 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..32b3b8e10a0 --- /dev/null +++ b/server/base/src/main/java/org/apache/accumulo/server/metadata/RefreshesImpl.java @@ -0,0 +1,237 @@ +/* + * 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.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()); + } + + // 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(1, 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 == 1); + 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 RefreshEntry decode(Map.Entry entry) { + String row = entry.getKey().getRowData().toString(); + Preconditions.checkArgument(row.startsWith(RefreshSection.getRowPrefix())); + Preconditions.checkArgument(entry.getKey().getColumnFamilyData().length() == 0); + Preconditions.checkArgument(entry.getKey().getColumnQualifierData().length() == 0); + + try (ByteArrayInputStream bais = new ByteArrayInputStream(entry.getValue().get()); + DataInputStream dis = new DataInputStream(bais)) { + Preconditions.checkArgument(dis.readInt() == 1); + 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 + Preconditions + .checkArgument(entries.stream().allMatch(e -> e.getExtent().equals(RootTable.EXTENT))); + 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 + Preconditions + .checkArgument(entries.stream().allMatch(e -> e.getExtent().equals(RootTable.EXTENT))); + 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 list() { + 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().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..e6b8a9f20be 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) { + + // process batches of refresh entries to avoid reading all into memory at once + Iterators.partition(ctx.getAmple().refreshes(dataLevel).list().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(); + ctx.getAmple().readTablets().forTablets(extents, Optional.empty()) + .fetch(PREV_ROW, LOCATION, SCANS).build().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..39217d4cab0 --- /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.list().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.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); + assertEquals(Map.of(ecid1, tserver1), + refreshes.list().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.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); + assertEquals(Map.of(ecid1, tserver1, ecid2, tserver1, ecid3, tserver2), + refreshes.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getTserver))); + + refreshes.delete(List.of(new RefreshEntry(ecid2, extent2, tserver1))); + + assertEquals(Map.of(ecid1, extent1, ecid3, extent2), + refreshes.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); + assertEquals(Map.of(ecid1, tserver1, ecid3, tserver2), + refreshes.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getTserver))); + + refreshes.delete(List.of(new RefreshEntry(ecid3, extent2, tserver2), + new RefreshEntry(ecid1, extent1, tserver1))); + + assertEquals(0, refreshes.list().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..678802cf354 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 copmactions 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(); } } From dedaa8890299b9a0222ee7dca88b591bb8012297 Mon Sep 17 00:00:00 2001 From: Keith Turner Date: Mon, 3 Jul 2023 15:18:46 -0400 Subject: [PATCH 2/4] code review updates --- .../accumulo/core/metadata/schema/Ample.java | 2 +- .../server/metadata/RefreshesImpl.java | 41 +++++++--- .../coordinator/CompactionCoordinator.java | 74 +++++++++---------- .../accumulo/test/compaction/RefreshesIT.java | 16 ++-- 4 files changed, 75 insertions(+), 58 deletions(-) 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 09f2e40aed6..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 @@ -625,7 +625,7 @@ public TServerInstance getTserver() { void delete(Collection entries); - Stream list(); + Stream stream(); } /** 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 index 32b3b8e10a0..531136853b4 100644 --- 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 @@ -32,6 +32,7 @@ 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; @@ -68,9 +69,13 @@ 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; @@ -82,13 +87,14 @@ public Data(int version, Map entries) { } private static String toJson(Map entries) { - var data = new Data(1, Objects.requireNonNull(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 == 1); + Preconditions.checkArgument(data.version == CURRENT_VERSION, "Expected version %s saw %s", + CURRENT_VERSION, data.version); Objects.requireNonNull(data.entries); return data.entries; } @@ -146,15 +152,28 @@ private Mutation createDeleteMutation(RefreshEntry entry) { 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())); - Preconditions.checkArgument(entry.getKey().getColumnFamilyData().length() == 0); - Preconditions.checkArgument(entry.getKey().getColumnQualifierData().length() == 0); + 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)) { - Preconditions.checkArgument(dis.readInt() == 1); + 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( @@ -170,8 +189,7 @@ 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 - Preconditions - .checkArgument(entries.stream().allMatch(e -> e.getExtent().equals(RootTable.EXTENT))); + requireRootTablets(entries); Consumer> mutator = map -> entries.forEach(refreshEntry -> map .put(refreshEntry.getEcid().canonical(), refreshEntry.getTserver().getHostPortSession())); mutateRootRefreshes(mutator); @@ -191,8 +209,7 @@ 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 - Preconditions - .checkArgument(entries.stream().allMatch(e -> e.getExtent().equals(RootTable.EXTENT))); + requireRootTablets(entries); Consumer> mutator = map -> entries.forEach(refreshEntry -> map.remove(refreshEntry.getEcid().canonical())); mutateRootRefreshes(mutator); @@ -208,7 +225,7 @@ public void delete(Collection entries) { } @Override - public Stream list() { + public Stream stream() { if (dataLevel == Ample.DataLevel.ROOT) { var zooReader = context.getZooReader(); byte[] jsonBytes; @@ -231,7 +248,7 @@ public Stream list() { throw new IllegalStateException(e); } scanner.setRange(range); - return scanner.stream().map(this::decode); + return scanner.stream().onClose(scanner::close).map(this::decode); } } } 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 e6b8a9f20be..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 @@ -199,47 +199,47 @@ protected void startRunningCleaner(ScheduledThreadPoolExecutor schedExecutor) { } 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)); + } - // process batches of refresh entries to avoid reading all into memory at once - Iterators.partition(ctx.getAmple().refreshes(dataLevel).list().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(); - ctx.getAmple().readTablets().forTablets(extents, Optional.empty()) - .fetch(PREV_ROW, LOCATION, SCANS).build().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(); - } + var tserverRefreshes = new HashMap>(); - ctx.getAmple().refreshes(dataLevel).delete(refreshEntries); + 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(); } 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 index 39217d4cab0..053f01d5a36 100644 --- a/test/src/main/java/org/apache/accumulo/test/compaction/RefreshesIT.java +++ b/test/src/main/java/org/apache/accumulo/test/compaction/RefreshesIT.java @@ -45,7 +45,7 @@ public class RefreshesIT extends ConfigurableMacBase { private void testRefreshes(Ample.DataLevel level, KeyExtent extent1, KeyExtent extent2) { var refreshes = getServerContext().getAmple().refreshes(level); - assertEquals(0, refreshes.list().count()); + assertEquals(0, refreshes.stream().count()); var ecid1 = ExternalCompactionId.generate(UUID.randomUUID()); var ecid2 = ExternalCompactionId.generate(UUID.randomUUID()); @@ -57,29 +57,29 @@ private void testRefreshes(Ample.DataLevel level, KeyExtent extent1, KeyExtent e refreshes.add(List.of(new RefreshEntry(ecid1, extent1, tserver1))); assertEquals(Map.of(ecid1, extent1), - refreshes.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); + refreshes.stream().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); assertEquals(Map.of(ecid1, tserver1), - refreshes.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getTserver))); + 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.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); + refreshes.stream().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); assertEquals(Map.of(ecid1, tserver1, ecid2, tserver1, ecid3, tserver2), - refreshes.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getTserver))); + 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.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); + refreshes.stream().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getExtent))); assertEquals(Map.of(ecid1, tserver1, ecid3, tserver2), - refreshes.list().collect(toMap(RefreshEntry::getEcid, RefreshEntry::getTserver))); + 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.list().count()); + assertEquals(0, refreshes.stream().count()); } @Test From 2401d2dc68260724449493be8d1455ffa8c099ec Mon Sep 17 00:00:00 2001 From: Keith Turner Date: Wed, 5 Jul 2023 11:26:56 -0400 Subject: [PATCH 3/4] Update test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java Co-authored-by: Dave Marion --- .../java/org/apache/accumulo/test/functional/CompactionIT.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 678802cf354..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 @@ -775,7 +775,7 @@ public void testSystemCompactionsRefresh() throws Exception { client.tableOperations().flush(tableName, null, null, true); - // there should be no system copmactions yet and no data should be filtered, so should see all + // 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()); From 0c844dafa0e7950c39c4a08df8f7332c29601cbc Mon Sep 17 00:00:00 2001 From: Keith Turner Date: Wed, 5 Jul 2023 15:04:33 -0400 Subject: [PATCH 4/4] fixed comment --- .../main/java/org/apache/accumulo/core/metadata/RootTable.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 6398b44283e..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 @@ -44,7 +44,7 @@ public class RootTable { */ public static final String ZROOT_TABLET_GC_CANDIDATES = ZROOT_TABLET + "/gc_candidates"; /* - * ZK path relative to the zookeeper node where the root tablet gc candidates are stored. + * 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";