You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2023/09/27 17:51:48 UTC
[accumulo] branch elasticity updated (83d2dc24e8 -> 8cd7c59412)
This is an automated email from the ASF dual-hosted git repository.
kturner pushed a change to branch elasticity
in repository https://gitbox.apache.org/repos/asf/accumulo.git
from 83d2dc24e8 Merge branch 'main' into elasticity
add 75140d8965 Add Message Prefix for File deletion actions in GCRun (#3745)
add f7e59d3653 Merge branch '2.1'
add 6c34c848c4 reverts #3747 that caused tablet split to fail (#3760)
add 48d69f30a1 Merge branch '2.1'
add 19719b5ff9 add check to gc to validate metadata scan is complete (#3703)
add 50914c4f52 Merge remote-tracking branch 'upstream/1.10' into 2.1
add cac52251fd Gc meta validation 2 1 (#3755)
add 7005006620 Merge remote-tracking branch 'upstream/2.1'
add 6dfc6ea4a4 Monitor resources version bump and cleanup (#3754)
add 2262125874 Remove popper changes to LICENSE files
add ff916e6759 Merge branch '2.1'
add 55d162046c Remove GC candidates found to still be referenced (#3738)
add b34a521efa Merge branch '2.1'
add 44211efc3c avoids creating server side threads when canceling failed write (#3756)
add f501dca59a Merge branch '2.1'
add 1ab0cc7808 No-chop merge and delete (#3761)
add af47a2f0bd Bump org.xerial.snappy:snappy-java from 1.1.10.1 to 1.1.10.4 (#3777)
add 5aa53a8e0c Merge branch '2.1'
add bcdadae248 throws exception when unknown compaction kind seen (#3778)
add 05950cd855 Update no-chop merge tests to disable compaction (#3775)
add aa32e15361 Remove unneeded chop compaction and split code (#3773)
add edfc2e2011 formats UpgraderConstants.java
add 6bf3aeec27 removes CHOP from CompactionKind SPI type. (#3781)
new 8cd7c59412 Merge branch 'main' into elasticity
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
LICENSE | 13 +-
.../core/client/admin/ActiveCompaction.java | 4 +-
.../apache/accumulo/core/client/rfile/RFile.java | 35 +
.../accumulo/core/client/rfile/RFileScanner.java | 2 +-
.../core/client/rfile/RFileScannerBuilder.java | 29 +-
.../accumulo/core/client/rfile/RFileSource.java | 23 +-
.../core/clientImpl/TabletServerBatchWriter.java | 46 +-
.../org/apache/accumulo/core/conf/Property.java | 5 +
.../org/apache/accumulo/core/file/rfile/RFile.java | 238 +-
.../accumulo/core/file/rfile/RFileOperations.java | 9 +-
.../org/apache/accumulo/core/gc/GcCandidate.java | 49 +-
.../org/apache/accumulo/core/gc/Reference.java | 7 +-
.../accumulo/core/gc/ReferenceDirectory.java | 8 +-
.../org/apache/accumulo/core/gc/ReferenceFile.java | 38 +-
.../core/manager/state/TabletManagement.java | 8 +-
.../accumulo/core/metadata/AbstractTabletFile.java | 42 +-
.../core/metadata/CompactableFileImpl.java | 3 +-
.../core/metadata/ReferencedTabletFile.java | 46 +-
.../accumulo/core/metadata/StoredTabletFile.java | 156 +-
.../apache/accumulo/core/metadata/TabletFile.java | 12 +
.../core/metadata/UnreferencedTabletFile.java | 15 +-
.../accumulo/core/metadata/ValidationUtil.java | 2 +-
.../accumulo/core/metadata/schema/Ample.java | 29 +-
.../schema/ExternalCompactionMetadata.java | 7 +-
.../core/metadata/schema/MetadataSchema.java | 10 -
.../core/metadata/schema/SelectedFiles.java | 2 +-
.../core/metadata/schema/TabletMetadata.java | 13 +-
.../metadata/schema/TabletMetadataBuilder.java | 8 -
.../core/metadata/schema/TabletMutatorBase.java | 24 +-
.../core/metadata/schema/TabletsMetadata.java | 4 -
.../core/spi/compaction/CompactionKind.java | 6 +-
.../core/spi/compaction/CompactionPlanner.java | 22 +-
.../spi/compaction/DefaultCompactionPlanner.java | 3 +-
.../util/compaction/CompactionJobPrioritizer.java | 1 -
.../util/json/ByteArrayToBase64TypeAdapter.java | 4 +-
.../core/manager/thrift/TabletLoadState.java | 5 +-
.../thrift/TabletIngestClientService.java | 1025 +++
.../core/tabletserver/thrift/TCompactionKind.java | 13 +-
.../tabletserver/thrift/TCompactionReason.java | 9 +-
core/src/main/thrift/manager.thrift | 1 -
core/src/main/thrift/tabletingest.thrift | 5 +
core/src/main/thrift/tabletserver.thrift | 2 -
.../core/client/rfile/RFileClientTest.java | 112 +-
.../core/file/rfile/AbstractRFileTest.java | 264 +
.../accumulo/core/file/rfile/FencedRFileTest.java | 376 +
.../core/file/rfile/MultiThreadedRFileTest.java | 2 +-
.../apache/accumulo/core/file/rfile/RFileTest.java | 176 +-
.../iterators/user/IndexedDocIteratorTest.java | 2 +-
.../metadata/schema/ReferencedTabletFileTest.java | 57 +-
.../core/metadata/schema/SelectedFilesTest.java | 11 +-
.../core/metadata/schema/TabletMetadataTest.java | 35 +-
.../util/compaction/CompactionPrioritizerTest.java | 4 +-
pom.xml | 2 +-
.../accumulo/server/compaction/CompactionInfo.java | 7 +-
.../accumulo/server/compaction/FileCompactor.java | 4 +-
.../server/constraints/MetadataConstraints.java | 34 +-
.../org/apache/accumulo/server/fs/VolumeUtil.java | 20 +-
.../accumulo/server/gc/AllVolumesDirectory.java | 4 +-
.../accumulo/server/init/ZooKeeperInitializer.java | 5 +-
.../accumulo/server/manager/state/MergeInfo.java | 9 +-
.../accumulo/server/manager/state/MergeState.java | 8 -
.../manager/state/TabletManagementIterator.java | 2 -
.../metadata/ConditionalTabletMutatorImpl.java | 5 +-
.../accumulo/server/metadata/RootGcCandidates.java | 23 +-
.../accumulo/server/metadata/ServerAmpleImpl.java | 32 +-
.../server/tablets/TabletNameGenerator.java | 18 +-
.../accumulo/server/util/ListVolumesUsed.java | 5 +-
.../accumulo/server/util/MetadataTableUtil.java | 14 +-
.../server/util/RemoveEntriesForMissingFiles.java | 4 +-
.../constraints/MetadataConstraintsTest.java | 171 +-
.../apache/accumulo/server/fs/VolumeUtilTest.java | 107 +-
.../server/manager/state/MergeInfoTest.java | 20 -
.../server/manager/state/TabletManagementTest.java | 36 +-
.../accumulo/server/util/TableDiskUsageTest.java | 10 +-
.../org/apache/accumulo/compactor/ExtCEnv.java | 5 +-
.../main/java/org/apache/accumulo/gc/GCRun.java | 116 +-
.../accumulo/gc/GarbageCollectionAlgorithm.java | 59 +-
.../accumulo/gc/GarbageCollectionEnvironment.java | 24 +-
.../apache/accumulo/gc/GarbageCollectionTest.java | 480 +-
.../accumulo/gc/SimpleGarbageCollectorTest.java | 58 +-
.../java/org/apache/accumulo/manager/Manager.java | 13 -
.../manager/ManagerClientServiceHandler.java | 3 -
.../accumulo/manager/TabletGroupWatcher.java | 518 +-
.../coordinator/CompactionCoordinator.java | 4 +-
.../accumulo/manager/recovery/RecoveryManager.java | 2 +-
.../apache/accumulo/manager/state/MergeStats.java | 93 +-
.../tableOps/bulkVer2/CleanUpBulkImport.java | 2 +-
.../tableOps/tableExport/WriteExportFiles.java | 5 +-
.../tableImport/PopulateMetadataTable.java | 9 +-
.../manager/upgrade/UpgraderConstants.java | 29 +-
.../accumulo/manager/TabletGroupWatcherTest.java | 67 +
.../accumulo/manager/split/SplitterTest.java | 27 +-
.../manager/tableOps/split/UpdateTabletsTest.java | 7 +-
.../src/main/appended-resources/META-INF/LICENSE | 13 +-
.../resources/external/bootstrap/css/bootstrap.css | 7217 ++++++++++++--------
.../external/bootstrap/css/bootstrap.css.map | 1 -
.../external/bootstrap/js/bootstrap.bundle.js | 6314 +++++++++++++++++
.../resources/external/bootstrap/js/bootstrap.js | 5016 --------------
.../external/bootstrap/js/bootstrap.js.map | 1 -
.../datatables/css/dataTables.bootstrap5.css | 87 +-
.../external/datatables/css/jquery.dataTables.css | 578 --
.../external/datatables/images/sort_asc.png | Bin 160 -> 0 bytes
.../datatables/images/sort_asc_disabled.png | Bin 148 -> 0 bytes
.../external/datatables/images/sort_both.png | Bin 201 -> 0 bytes
.../external/datatables/images/sort_desc.png | Bin 158 -> 0 bytes
.../datatables/images/sort_desc_disabled.png | Bin 146 -> 0 bytes
.../datatables/js/dataTables.bootstrap5.js | 86 +-
.../external/datatables/js/jquery.dataTables.js | 235 +-
.../external/flot/jquery.flot.drawSeries.js | 123 +-
.../jquery/{jquery-3.6.1.js => jquery-3.7.1.js} | 1915 +++---
.../apache/accumulo/monitor/templates/default.ftl | 4 +-
.../accumulo/tserver/TabletClientHandler.java | 5 +
.../accumulo/tserver/session/SessionManager.java | 25 +
.../tserver/tablet/MinorCompactionTask.java | 2 +-
.../accumulo/tserver/tablet/MinorCompactor.java | 2 +-
.../org/apache/accumulo/tserver/tablet/Tablet.java | 6 +-
.../tserver/compaction/CompactableUtilsTest.java | 2 +-
.../shell/commands/ListTabletsCommandTest.java | 14 +-
.../java/org/apache/accumulo/test/CloneIT.java | 41 +-
.../java/org/apache/accumulo/test/GCRunIT.java | 201 +
.../accumulo/test/ScanServerMetadataEntriesIT.java | 4 +-
.../java/org/apache/accumulo/test/VolumeIT.java | 16 +-
.../accumulo/test/functional/BulkFailureIT.java | 3 +-
.../apache/accumulo/test/functional/BulkNewIT.java | 2 +-
.../accumulo/test/functional/CleanTmpIT.java | 3 +-
.../accumulo/test/functional/CloneTestIT.java | 3 +-
.../accumulo/test/functional/DeleteRowsIT.java | 102 +-
.../accumulo/test/functional/FileMetadataIT.java | 488 ++
.../test/functional/FileNormalizationIT.java | 10 +-
.../test/functional/FunctionalTestUtils.java | 14 +-
.../test/functional/GarbageCollectorIT.java | 195 +-
.../apache/accumulo/test/functional/MergeIT.java | 345 +-
.../accumulo/test/functional/ReadWriteIT.java | 3 +-
.../test/functional/RecoveryWithEmptyRFileIT.java | 3 +-
.../accumulo/test/functional/SplitRecoveryIT.java | 4 +-
.../apache/accumulo/test/manager/MergeStateIT.java | 16 +-
.../accumulo/test/performance/NullTserver.java | 5 +
.../accumulo/test/util/FileMetadataUtil.java | 76 +
138 files changed, 17262 insertions(+), 11001 deletions(-)
copy minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/ProcessReference.java => core/src/main/java/org/apache/accumulo/core/gc/GcCandidate.java (55%)
create mode 100644 core/src/test/java/org/apache/accumulo/core/file/rfile/AbstractRFileTest.java
create mode 100644 core/src/test/java/org/apache/accumulo/core/file/rfile/FencedRFileTest.java
copy core/src/main/java/org/apache/accumulo/core/data/TabletId.java => server/manager/src/main/java/org/apache/accumulo/manager/upgrade/UpgraderConstants.java (63%)
create mode 100644 server/manager/src/test/java/org/apache/accumulo/manager/TabletGroupWatcherTest.java
delete mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/bootstrap/css/bootstrap.css.map
create mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/bootstrap/js/bootstrap.bundle.js
delete mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/bootstrap/js/bootstrap.js
delete mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/bootstrap/js/bootstrap.js.map
delete mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/datatables/css/jquery.dataTables.css
delete mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/datatables/images/sort_asc.png
delete mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/datatables/images/sort_asc_disabled.png
delete mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/datatables/images/sort_both.png
delete mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/datatables/images/sort_desc.png
delete mode 100644 server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/datatables/images/sort_desc_disabled.png
rename server/monitor/src/main/resources/org/apache/accumulo/monitor/resources/external/jquery/{jquery-3.6.1.js => jquery-3.7.1.js} (88%)
create mode 100644 test/src/main/java/org/apache/accumulo/test/GCRunIT.java
create mode 100644 test/src/main/java/org/apache/accumulo/test/functional/FileMetadataIT.java
create mode 100644 test/src/main/java/org/apache/accumulo/test/util/FileMetadataUtil.java
[accumulo] 01/01: Merge branch 'main' into elasticity
Posted by kt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kturner pushed a commit to branch elasticity
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 8cd7c59412887b4d06b40610e51412293fca870d
Merge: 83d2dc24e8 6bf3aeec27
Author: Keith Turner <kt...@apache.org>
AuthorDate: Wed Sep 27 13:45:27 2023 -0400
Merge branch 'main' into elasticity
LICENSE | 13 +-
.../core/client/admin/ActiveCompaction.java | 4 +-
.../apache/accumulo/core/client/rfile/RFile.java | 35 +
.../accumulo/core/client/rfile/RFileScanner.java | 2 +-
.../core/client/rfile/RFileScannerBuilder.java | 29 +-
.../accumulo/core/client/rfile/RFileSource.java | 23 +-
.../core/clientImpl/TabletServerBatchWriter.java | 46 +-
.../org/apache/accumulo/core/conf/Property.java | 5 +
.../org/apache/accumulo/core/file/rfile/RFile.java | 238 +-
.../accumulo/core/file/rfile/RFileOperations.java | 9 +-
.../org/apache/accumulo/core/gc/GcCandidate.java | 72 +
.../org/apache/accumulo/core/gc/Reference.java | 7 +-
.../accumulo/core/gc/ReferenceDirectory.java | 8 +-
.../org/apache/accumulo/core/gc/ReferenceFile.java | 38 +-
.../core/manager/state/TabletManagement.java | 8 +-
.../accumulo/core/metadata/AbstractTabletFile.java | 42 +-
.../core/metadata/CompactableFileImpl.java | 3 +-
.../core/metadata/ReferencedTabletFile.java | 46 +-
.../accumulo/core/metadata/StoredTabletFile.java | 156 +-
.../apache/accumulo/core/metadata/TabletFile.java | 12 +
.../core/metadata/UnreferencedTabletFile.java | 15 +-
.../accumulo/core/metadata/ValidationUtil.java | 2 +-
.../accumulo/core/metadata/schema/Ample.java | 29 +-
.../schema/ExternalCompactionMetadata.java | 7 +-
.../core/metadata/schema/MetadataSchema.java | 10 -
.../core/metadata/schema/SelectedFiles.java | 2 +-
.../core/metadata/schema/TabletMetadata.java | 13 +-
.../metadata/schema/TabletMetadataBuilder.java | 8 -
.../core/metadata/schema/TabletMutatorBase.java | 24 +-
.../core/metadata/schema/TabletsMetadata.java | 4 -
.../core/spi/compaction/CompactionKind.java | 6 +-
.../core/spi/compaction/CompactionPlanner.java | 22 +-
.../spi/compaction/DefaultCompactionPlanner.java | 3 +-
.../util/compaction/CompactionJobPrioritizer.java | 1 -
.../util/json/ByteArrayToBase64TypeAdapter.java | 4 +-
.../core/manager/thrift/TabletLoadState.java | 5 +-
.../thrift/TabletIngestClientService.java | 1025 +++
.../core/tabletserver/thrift/TCompactionKind.java | 13 +-
.../tabletserver/thrift/TCompactionReason.java | 9 +-
core/src/main/thrift/manager.thrift | 1 -
core/src/main/thrift/tabletingest.thrift | 5 +
core/src/main/thrift/tabletserver.thrift | 2 -
.../core/client/rfile/RFileClientTest.java | 112 +-
.../core/file/rfile/AbstractRFileTest.java | 264 +
.../accumulo/core/file/rfile/FencedRFileTest.java | 376 +
.../core/file/rfile/MultiThreadedRFileTest.java | 2 +-
.../apache/accumulo/core/file/rfile/RFileTest.java | 176 +-
.../iterators/user/IndexedDocIteratorTest.java | 2 +-
.../metadata/schema/ReferencedTabletFileTest.java | 57 +-
.../core/metadata/schema/SelectedFilesTest.java | 11 +-
.../core/metadata/schema/TabletMetadataTest.java | 35 +-
.../util/compaction/CompactionPrioritizerTest.java | 4 +-
pom.xml | 2 +-
.../accumulo/server/compaction/CompactionInfo.java | 7 +-
.../accumulo/server/compaction/FileCompactor.java | 4 +-
.../server/constraints/MetadataConstraints.java | 34 +-
.../org/apache/accumulo/server/fs/VolumeUtil.java | 20 +-
.../accumulo/server/gc/AllVolumesDirectory.java | 4 +-
.../accumulo/server/init/ZooKeeperInitializer.java | 5 +-
.../accumulo/server/manager/state/MergeInfo.java | 9 +-
.../accumulo/server/manager/state/MergeState.java | 8 -
.../manager/state/TabletManagementIterator.java | 2 -
.../metadata/ConditionalTabletMutatorImpl.java | 5 +-
.../accumulo/server/metadata/RootGcCandidates.java | 23 +-
.../accumulo/server/metadata/ServerAmpleImpl.java | 32 +-
.../server/tablets/TabletNameGenerator.java | 18 +-
.../accumulo/server/util/ListVolumesUsed.java | 5 +-
.../accumulo/server/util/MetadataTableUtil.java | 14 +-
.../server/util/RemoveEntriesForMissingFiles.java | 4 +-
.../constraints/MetadataConstraintsTest.java | 171 +-
.../apache/accumulo/server/fs/VolumeUtilTest.java | 107 +-
.../server/manager/state/MergeInfoTest.java | 20 -
.../server/manager/state/TabletManagementTest.java | 36 +-
.../accumulo/server/util/TableDiskUsageTest.java | 10 +-
.../org/apache/accumulo/compactor/ExtCEnv.java | 5 +-
.../main/java/org/apache/accumulo/gc/GCRun.java | 116 +-
.../accumulo/gc/GarbageCollectionAlgorithm.java | 59 +-
.../accumulo/gc/GarbageCollectionEnvironment.java | 24 +-
.../apache/accumulo/gc/GarbageCollectionTest.java | 480 +-
.../accumulo/gc/SimpleGarbageCollectorTest.java | 58 +-
.../java/org/apache/accumulo/manager/Manager.java | 13 -
.../manager/ManagerClientServiceHandler.java | 3 -
.../accumulo/manager/TabletGroupWatcher.java | 518 +-
.../coordinator/CompactionCoordinator.java | 4 +-
.../accumulo/manager/recovery/RecoveryManager.java | 2 +-
.../apache/accumulo/manager/state/MergeStats.java | 93 +-
.../tableOps/bulkVer2/CleanUpBulkImport.java | 2 +-
.../tableOps/tableExport/WriteExportFiles.java | 5 +-
.../tableImport/PopulateMetadataTable.java | 9 +-
.../manager/upgrade/UpgraderConstants.java | 33 +-
.../accumulo/manager/TabletGroupWatcherTest.java | 67 +
.../accumulo/manager/split/SplitterTest.java | 27 +-
.../manager/tableOps/split/UpdateTabletsTest.java | 7 +-
.../src/main/appended-resources/META-INF/LICENSE | 13 +-
.../resources/external/bootstrap/css/bootstrap.css | 7217 ++++++++++++--------
.../external/bootstrap/css/bootstrap.css.map | 1 -
.../external/bootstrap/js/bootstrap.bundle.js | 6314 +++++++++++++++++
.../resources/external/bootstrap/js/bootstrap.js | 5016 --------------
.../external/bootstrap/js/bootstrap.js.map | 1 -
.../datatables/css/dataTables.bootstrap5.css | 87 +-
.../external/datatables/css/jquery.dataTables.css | 578 --
.../external/datatables/images/sort_asc.png | Bin 160 -> 0 bytes
.../datatables/images/sort_asc_disabled.png | Bin 148 -> 0 bytes
.../external/datatables/images/sort_both.png | Bin 201 -> 0 bytes
.../external/datatables/images/sort_desc.png | Bin 158 -> 0 bytes
.../datatables/images/sort_desc_disabled.png | Bin 146 -> 0 bytes
.../datatables/js/dataTables.bootstrap5.js | 86 +-
.../external/datatables/js/jquery.dataTables.js | 235 +-
.../external/flot/jquery.flot.drawSeries.js | 123 +-
.../jquery/{jquery-3.6.1.js => jquery-3.7.1.js} | 1915 +++---
.../apache/accumulo/monitor/templates/default.ftl | 4 +-
.../accumulo/tserver/TabletClientHandler.java | 5 +
.../accumulo/tserver/session/SessionManager.java | 25 +
.../tserver/tablet/MinorCompactionTask.java | 2 +-
.../accumulo/tserver/tablet/MinorCompactor.java | 2 +-
.../org/apache/accumulo/tserver/tablet/Tablet.java | 6 +-
.../tserver/compaction/CompactableUtilsTest.java | 2 +-
.../shell/commands/ListTabletsCommandTest.java | 14 +-
.../java/org/apache/accumulo/test/CloneIT.java | 41 +-
.../java/org/apache/accumulo/test/GCRunIT.java | 201 +
.../accumulo/test/ScanServerMetadataEntriesIT.java | 4 +-
.../java/org/apache/accumulo/test/VolumeIT.java | 16 +-
.../accumulo/test/functional/BulkFailureIT.java | 3 +-
.../apache/accumulo/test/functional/BulkNewIT.java | 2 +-
.../accumulo/test/functional/CleanTmpIT.java | 3 +-
.../accumulo/test/functional/CloneTestIT.java | 3 +-
.../accumulo/test/functional/DeleteRowsIT.java | 102 +-
.../accumulo/test/functional/FileMetadataIT.java | 488 ++
.../test/functional/FileNormalizationIT.java | 10 +-
.../test/functional/FunctionalTestUtils.java | 14 +-
.../test/functional/GarbageCollectorIT.java | 195 +-
.../apache/accumulo/test/functional/MergeIT.java | 345 +-
.../accumulo/test/functional/ReadWriteIT.java | 3 +-
.../test/functional/RecoveryWithEmptyRFileIT.java | 3 +-
.../accumulo/test/functional/SplitRecoveryIT.java | 4 +-
.../apache/accumulo/test/manager/MergeStateIT.java | 16 +-
.../accumulo/test/performance/NullTserver.java | 5 +
.../accumulo/test/util/FileMetadataUtil.java | 76 +
138 files changed, 17304 insertions(+), 10986 deletions(-)
diff --cc core/src/main/java/org/apache/accumulo/core/manager/state/TabletManagement.java
index e99b647a50,0000000000..a687de873f
mode 100644,000000..100644
--- a/core/src/main/java/org/apache/accumulo/core/manager/state/TabletManagement.java
+++ b/core/src/main/java/org/apache/accumulo/core/manager/state/TabletManagement.java
@@@ -1,101 -1,0 +1,101 @@@
+/*
+ * 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.core.manager.state;
+
+import java.io.IOException;
+import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.SortedMap;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.user.WholeRowIterator;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
+import org.apache.hadoop.io.Text;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+
+/**
+ * Object that represents a Tablets metadata and any actions that the Manager might need to take on
+ * the object. This object is created by the TabletManagementIterator iterator used by the
+ * TabletGroupWatcher threads in the Manager.
+ *
+ */
+public class TabletManagement {
+
- public static final EnumSet<ColumnType> CONFIGURED_COLUMNS = EnumSet.of(ColumnType.PREV_ROW,
- ColumnType.LOCATION, ColumnType.SUSPEND, ColumnType.LOGS, ColumnType.CHOPPED,
- ColumnType.HOSTING_GOAL, ColumnType.HOSTING_REQUESTED, ColumnType.FILES, ColumnType.LAST,
- ColumnType.OPID, ColumnType.ECOMP, ColumnType.DIR, ColumnType.SELECTED);
++ public static final EnumSet<ColumnType> CONFIGURED_COLUMNS =
++ EnumSet.of(ColumnType.PREV_ROW, ColumnType.LOCATION, ColumnType.SUSPEND, ColumnType.LOGS,
++ ColumnType.HOSTING_GOAL, ColumnType.HOSTING_REQUESTED, ColumnType.FILES, ColumnType.LAST,
++ ColumnType.OPID, ColumnType.ECOMP, ColumnType.DIR, ColumnType.SELECTED);
+
+ private static final Text REASONS_COLUMN_NAME = new Text("REASONS");
+
+ private static final Text EMPTY = new Text("");
+
+ public static enum ManagementAction {
+ BAD_STATE, NEEDS_COMPACTING, NEEDS_LOCATION_UPDATE, IS_MERGING, NEEDS_SPLITTING;
+ }
+
+ public static void addActions(final SortedMap<Key,Value> decodedRow,
+ final Set<ManagementAction> actions) {
+ final Key reasonsKey = new Key(decodedRow.firstKey().getRow(), REASONS_COLUMN_NAME, EMPTY);
+ final Value reasonsValue = new Value(Joiner.on(',').join(actions));
+ decodedRow.put(reasonsKey, reasonsValue);
+ }
+
+ public final Set<ManagementAction> actions;
+ public final TabletMetadata tabletMetadata;
+
+ public TabletManagement(Set<ManagementAction> actions, TabletMetadata tm) {
+ this.actions = actions;
+ this.tabletMetadata = tm;
+ }
+
+ public TabletManagement(Key wholeRowKey, Value wholeRowValue) throws IOException {
+ this(wholeRowKey, wholeRowValue, false);
+ }
+
+ public TabletManagement(Key wholeRowKey, Value wholeRowValue, boolean saveKV) throws IOException {
+ final SortedMap<Key,Value> decodedRow = WholeRowIterator.decodeRow(wholeRowKey, wholeRowValue);
+ Text row = decodedRow.firstKey().getRow();
+ Value val = decodedRow.remove(new Key(row, REASONS_COLUMN_NAME, EMPTY));
+ Set<ManagementAction> actions = new HashSet<>();
+ Splitter.on(',').split(val.toString()).forEach(a -> actions.add(ManagementAction.valueOf(a)));
+ TabletMetadata tm = TabletMetadata.convertRow(decodedRow.entrySet().iterator(),
+ CONFIGURED_COLUMNS, saveKV, true);
+ this.actions = actions;
+ this.tabletMetadata = tm;
+ }
+
+ public Set<ManagementAction> getActions() {
+ return actions;
+ }
+
+ public TabletMetadata getTabletMetadata() {
+ return tabletMetadata;
+ }
+
+ @Override
+ public String toString() {
+ return actions.toString() + "," + tabletMetadata.toString();
+ }
+}
diff --cc core/src/main/java/org/apache/accumulo/core/metadata/CompactableFileImpl.java
index d09ad4e175,31fdc55d16..a8296ce495
--- a/core/src/main/java/org/apache/accumulo/core/metadata/CompactableFileImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/CompactableFileImpl.java
@@@ -30,9 -31,7 +31,9 @@@ public class CompactableFileImpl implem
private final DataFileValue dataFileValue;
public CompactableFileImpl(URI uri, long size, long entries) {
+ // TODO this normalizes the path passing it through URI defeating the purpose of
+ // StoredTabletFile
- this.storedTabletFile = new StoredTabletFile(uri.toString());
+ this.storedTabletFile = StoredTabletFile.of(uri, new Range());
this.dataFileValue = new DataFileValue(size, entries);
}
diff --cc core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java
index bf289f82e4,553f029e1f..af3506a4fb
--- 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
@@@ -290,73 -265,52 +313,71 @@@ public interface Ample
/**
* Interface for changing a tablets persistent data.
*/
- interface TabletMutator {
- TabletMutator putPrevEndRow(Text per);
+ interface TabletUpdates<T> {
+ T putPrevEndRow(Text per);
+
+ T putFile(ReferencedTabletFile path, DataFileValue dfv);
+
+ T putFile(StoredTabletFile path, DataFileValue dfv);
+
+ T deleteFile(StoredTabletFile path);
+
+ T putScan(StoredTabletFile path);
+
+ T deleteScan(StoredTabletFile path);
- TabletMutator putFile(ReferencedTabletFile path, DataFileValue dfv);
+ T putCompactionId(long compactionId);
- TabletMutator putFile(StoredTabletFile path, DataFileValue dfv);
+ T putFlushId(long flushId);
- TabletMutator deleteFile(StoredTabletFile path);
+ T putLocation(Location location);
- TabletMutator putScan(StoredTabletFile path);
+ T deleteLocation(Location location);
- TabletMutator deleteScan(StoredTabletFile path);
+ T putZooLock(String zookeeperRoot, ServiceLock zooLock);
- TabletMutator putCompactionId(long compactionId);
+ T putDirName(String dirName);
- TabletMutator putFlushId(long flushId);
+ T putWal(LogEntry logEntry);
- TabletMutator putLocation(Location location);
+ T deleteWal(String wal);
- TabletMutator deleteLocation(Location location);
+ T deleteWal(LogEntry logEntry);
- TabletMutator putZooLock(ServiceLock zooLock);
+ T putTime(MetadataTime time);
- TabletMutator putDirName(String dirName);
+ T putBulkFile(ReferencedTabletFile bulkref, long tid);
- TabletMutator putWal(LogEntry logEntry);
+ T deleteBulkFile(StoredTabletFile bulkref);
- T putChopped();
-
- TabletMutator deleteWal(String wal);
+ T putSuspension(TServerInstance tserver, long suspensionTime);
- TabletMutator deleteWal(LogEntry logEntry);
+ T deleteSuspension();
- TabletMutator putTime(MetadataTime time);
+ T putExternalCompaction(ExternalCompactionId ecid, ExternalCompactionMetadata ecMeta);
- TabletMutator putBulkFile(ReferencedTabletFile bulkref, long tid);
+ T deleteExternalCompaction(ExternalCompactionId ecid);
- TabletMutator deleteBulkFile(StoredTabletFile bulkref);
+ T putCompacted(long fateTxid);
- TabletMutator putSuspension(TServerInstance tserver, long suspensionTime);
+ T deleteCompacted(long fateTxid);
- TabletMutator deleteSuspension();
+ T putHostingGoal(TabletHostingGoal goal);
- TabletMutator putExternalCompaction(ExternalCompactionId ecid,
- ExternalCompactionMetadata ecMeta);
+ T setHostingRequested();
- TabletMutator deleteExternalCompaction(ExternalCompactionId ecid);
+ T deleteHostingRequested();
+ T putOperation(TabletOperationId opId);
+
+ T deleteOperation();
+
+ T putSelectedFiles(SelectedFiles selectedFiles);
+
+ T deleteSelectedFiles();
+ }
+
+ interface TabletMutator extends TabletUpdates<TabletMutator> {
/**
* This method persist (or queues for persisting) previous put and deletes against this object.
* Unless this method is called, previous calls will never be persisted. The purpose of this
diff --cc core/src/main/java/org/apache/accumulo/core/metadata/schema/ExternalCompactionMetadata.java
index 3a6d988216,a5beddd7ad..c6df9c8673
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/ExternalCompactionMetadata.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/ExternalCompactionMetadata.java
@@@ -105,8 -125,9 +104,8 @@@ public class ExternalCompactionMetadat
public String toJson() {
GSonData jData = new GSonData();
- jData.inputs = jobFiles.stream().map(StoredTabletFile::getMetaUpdateDelete).collect(toList());
- jData.tmp = compactTmpName.getMetaInsert();
+ jData.inputs = jobFiles.stream().map(StoredTabletFile::getMetadata).collect(toList());
- jData.nextFiles = nextFiles.stream().map(StoredTabletFile::getMetadata).collect(toList());
+ jData.tmp = compactTmpName.insert().getMetadata();
jData.compactor = compactorId;
jData.kind = kind.name();
jData.executorId = ((CompactionExecutorIdImpl) ceid).getExternalName();
@@@ -121,9 -143,11 +120,9 @@@
return new ExternalCompactionMetadata(
jData.inputs.stream().map(StoredTabletFile::new).collect(toSet()),
- new ReferencedTabletFile(new Path(jData.tmp)), jData.compactor,
- jData.nextFiles.stream().map(StoredTabletFile::new).collect(toSet()),
+ StoredTabletFile.of(jData.tmp).getTabletFile(), jData.compactor,
CompactionKind.valueOf(jData.kind), jData.priority,
- CompactionExecutorIdImpl.externalId(jData.executorId), jData.propDels, jData.selectedAll,
- jData.compactionId);
+ CompactionExecutorIdImpl.externalId(jData.executorId), jData.propDels, jData.fateTxId);
}
@Override
diff --cc core/src/main/java/org/apache/accumulo/core/metadata/schema/SelectedFiles.java
index 73f005199b,0000000000..fb50ef1063
mode 100644,000000..100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/SelectedFiles.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/SelectedFiles.java
@@@ -1,161 -1,0 +1,161 @@@
+/*
+ * 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.core.metadata.schema;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.accumulo.core.fate.FateTxId;
+import org.apache.accumulo.core.metadata.StoredTabletFile;
+
+import com.google.common.base.Preconditions;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.TypeAdapter;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonWriter;
+
+/**
+ * This class is used to manage the set of files selected for a user compaction for a tablet.
+ */
+public class SelectedFiles {
+
+ private final Set<StoredTabletFile> files;
+ private final boolean initiallySelectedAll;
+ private final long fateTxId;
+
+ private String metadataValue;
+
+ private static final Gson GSON = new GsonBuilder()
+ .registerTypeAdapter(SelectedFiles.class, new SelectedFilesTypeAdapter()).create();
+
+ public SelectedFiles(Set<StoredTabletFile> files, boolean initiallySelectedAll, long fateTxId) {
+ Preconditions.checkArgument(files != null && !files.isEmpty());
+ this.files = Set.copyOf(files);
+ this.initiallySelectedAll = initiallySelectedAll;
+ this.fateTxId = fateTxId;
+ }
+
+ private static class SelectedFilesTypeAdapter extends TypeAdapter<SelectedFiles> {
+
+ @Override
+ public void write(JsonWriter out, SelectedFiles selectedFiles) throws IOException {
+ out.beginObject();
+ out.name("txid").value(FateTxId.formatTid(selectedFiles.getFateTxId()));
+ out.name("selAll").value(selectedFiles.initiallySelectedAll());
+ out.name("files").beginArray();
+ // sort the data to make serialized json comparable
- selectedFiles.getFiles().stream().map(StoredTabletFile::getMetaUpdateDelete).sorted()
++ selectedFiles.getFiles().stream().map(StoredTabletFile::getMetadata).sorted()
+ .forEach(file -> {
+ try {
+ out.value(file);
+ } catch (IOException e) {
+ throw new UncheckedIOException(
+ "Failed to add file " + file + " to the JSON files array", e);
+ }
+ });
+ out.endArray();
+ out.endObject();
+ }
+
+ @Override
+ public SelectedFiles read(JsonReader in) throws IOException {
+ long fateTxId = 0L;
+ boolean selAll = false;
+ List<String> files = new ArrayList<>();
+
+ in.beginObject();
+ while (in.hasNext()) {
+ String name = in.nextName();
+ switch (name) {
+ case "txid":
+ fateTxId = FateTxId.fromString(in.nextString());
+ break;
+ case "selAll":
+ selAll = in.nextBoolean();
+ break;
+ case "files":
+ in.beginArray();
+ while (in.hasNext()) {
+ files.add(in.nextString());
+ }
+ in.endArray();
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown field name : " + name);
+ }
+ }
+ in.endObject();
+
+ Set<StoredTabletFile> tabletFiles =
+ files.stream().map(StoredTabletFile::new).collect(Collectors.toSet());
+
+ return new SelectedFiles(tabletFiles, selAll, fateTxId);
+ }
+
+ }
+
+ public static SelectedFiles from(String json) {
+ return GSON.fromJson(json, SelectedFiles.class);
+ }
+
+ public Set<StoredTabletFile> getFiles() {
+ return files;
+ }
+
+ public boolean initiallySelectedAll() {
+ return initiallySelectedAll;
+ }
+
+ public long getFateTxId() {
+ return fateTxId;
+ }
+
+ public String getMetadataValue() {
+ if (this.metadataValue == null) {
+ // use the custom TypeAdapter to create the json
+ this.metadataValue = GSON.toJson(this);
+ }
+ return this.metadataValue;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ SelectedFiles other = (SelectedFiles) obj;
+ return fateTxId == other.fateTxId && files.equals(other.files)
+ && initiallySelectedAll == other.initiallySelectedAll;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(fateTxId, files, initiallySelectedAll);
+ }
+
+}
diff --cc core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
index 31c1ffaf04,bc58b2b806..39eb11c97a
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
@@@ -61,10 -52,10 +61,9 @@@ import org.apache.accumulo.core.metadat
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.SuspendingTServer;
import org.apache.accumulo.core.metadata.TServerInstance;
-import org.apache.accumulo.core.metadata.TabletLocationState;
-import org.apache.accumulo.core.metadata.TabletState;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.BulkFileColumnFamily;
- import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ClonedColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CompactedColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CurrentLocationColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ExternalCompactionColumnFamily;
@@@ -119,17 -104,6 +118,16 @@@ public class TabletMetadata
private OptionalLong compact = OptionalLong.empty();
private Double splitRatio = null;
private Map<ExternalCompactionId,ExternalCompactionMetadata> extCompactions;
- private boolean chopped = false;
+ private TabletHostingGoal goal = TabletHostingGoal.ONDEMAND;
+ private boolean onDemandHostingRequested = false;
+ private TabletOperationId operationId;
+ private boolean futureAndCurrentLocationSet = false;
+ private boolean operationIdAndCurrentLocationSet = false;
+ private Set<Long> compacted;
+
+ public static TabletMetadataBuilder builder(KeyExtent extent) {
+ return new TabletMetadataBuilder(extent);
+ }
public enum LocationType {
CURRENT, FUTURE, LAST
@@@ -151,13 -125,7 +149,12 @@@
COMPACT_ID,
SPLIT_RATIO,
SUSPEND,
- CHOPPED,
- ECOMP
+ ECOMP,
+ HOSTING_GOAL,
+ HOSTING_REQUESTED,
+ OPID,
+ SELECTED,
+ COMPACTED
}
public static class Location {
@@@ -383,43 -345,6 +380,38 @@@
return splitRatio;
}
- public boolean hasChopped() {
- ensureFetched(ColumnType.CHOPPED);
- return chopped;
- }
-
+ public TabletHostingGoal getHostingGoal() {
+ if (RootTable.ID.equals(getTableId()) || MetadataTable.ID.equals(getTableId())) {
+ // Override the goal for the system tables
+ return TabletHostingGoal.ALWAYS;
+ }
+ ensureFetched(ColumnType.HOSTING_GOAL);
+ return goal;
+ }
+
+ public boolean getHostingRequested() {
+ ensureFetched(ColumnType.HOSTING_REQUESTED);
+ return onDemandHostingRequested;
+ }
+
+ @Override
+ public String toString() {
+ return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("tableId", tableId)
+ .append("prevEndRow", prevEndRow).append("sawPrevEndRow", sawPrevEndRow)
+ .append("oldPrevEndRow", oldPrevEndRow).append("sawOldPrevEndRow", sawOldPrevEndRow)
+ .append("endRow", endRow).append("location", location).append("files", files)
+ .append("scans", scans).append("loadedFiles", loadedFiles)
+ .append("fetchedCols", fetchedCols).append("extent", extent).append("last", last)
+ .append("suspend", suspend).append("dirName", dirName).append("time", time)
+ .append("cloned", cloned).append("flush", flush).append("logs", logs)
+ .append("compact", compact).append("splitRatio", splitRatio)
- .append("extCompactions", extCompactions).append("chopped", chopped).append("goal", goal)
++ .append("extCompactions", extCompactions).append("goal", goal)
+ .append("onDemandHostingRequested", onDemandHostingRequested)
+ .append("operationId", operationId).append("selectedFiles", selectedFiles)
+ .append("futureAndCurrentLocationSet", futureAndCurrentLocationSet)
+ .append("operationIdAndCurrentLocationSet", operationIdAndCurrentLocationSet).toString();
+ }
+
public SortedMap<Key,Value> getKeyValues() {
Preconditions.checkState(keyValues != null, "Requested key values when it was not saved");
return keyValues;
@@@ -563,24 -479,8 +555,21 @@@
extCompBuilder.put(ExternalCompactionId.of(qual),
ExternalCompactionMetadata.fromJson(val));
break;
- default:
- throw new IllegalStateException("Unexpected family " + fam);
+ case CompactedColumnFamily.STR_NAME:
+ compactedBuilder.add(FateTxId.fromString(qual));
+ break;
- case ChoppedColumnFamily.STR_NAME:
- te.chopped = true;
- break;
+ case HostingColumnFamily.STR_NAME:
+ switch (qual) {
+ case GOAL_QUAL:
+ te.goal = TabletHostingGoalUtil.fromValue(kv.getValue());
+ break;
+ case REQUESTED_QUAL:
+ te.onDemandHostingRequested = true;
+ break;
+ default:
+ throw new IllegalStateException("Unexpected family " + fam);
+ }
+ break;
}
}
diff --cc core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadataBuilder.java
index 305b5861fd,0000000000..7df5c34f26
mode 100644,000000..100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadataBuilder.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadataBuilder.java
@@@ -1,300 -1,0 +1,292 @@@
+/*
+ * 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.core.metadata.schema;
+
- import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.CHOPPED;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.COMPACTED;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.COMPACT_ID;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.DIR;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.ECOMP;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.FILES;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.FLUSH_ID;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.HOSTING_GOAL;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.HOSTING_REQUESTED;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOADED;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOCATION;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOGS;
+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 static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.SUSPEND;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.TIME;
+
+import java.util.Arrays;
+import java.util.EnumSet;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+import org.apache.accumulo.core.client.admin.TabletHostingGoal;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.lock.ServiceLock;
+import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.StoredTabletFile;
+import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
+import org.apache.hadoop.io.Text;
+
+public class TabletMetadataBuilder implements Ample.TabletUpdates<TabletMetadataBuilder> {
+
+ public static class InternalBuilder extends TabletMutatorBase<InternalBuilder> {
+ protected InternalBuilder(KeyExtent extent) {
+ super(extent);
+ }
+
+ @Override
+ public Mutation getMutation() {
+ return super.getMutation();
+ }
+ }
+
+ private final InternalBuilder internalBuilder;
+ EnumSet<TabletMetadata.ColumnType> fetched;
+
+ protected TabletMetadataBuilder(KeyExtent extent) {
+ internalBuilder = new InternalBuilder(extent);
+ fetched = EnumSet.noneOf(TabletMetadata.ColumnType.class);
+ putPrevEndRow(extent.prevEndRow());
+ }
+
+ @Override
+ public TabletMetadataBuilder putPrevEndRow(Text per) {
+ fetched.add(PREV_ROW);
+ internalBuilder.putPrevEndRow(per);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder putFile(ReferencedTabletFile path, DataFileValue dfv) {
+ fetched.add(FILES);
+ internalBuilder.putFile(path, dfv);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder putFile(StoredTabletFile path, DataFileValue dfv) {
+ fetched.add(FILES);
+ internalBuilder.putFile(path, dfv);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteFile(StoredTabletFile path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putScan(StoredTabletFile path) {
+ fetched.add(SCANS);
+ internalBuilder.putScan(path);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteScan(StoredTabletFile path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putCompactionId(long compactionId) {
+ fetched.add(COMPACT_ID);
+ internalBuilder.putCompactionId(compactionId);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder putFlushId(long flushId) {
+ fetched.add(FLUSH_ID);
+ internalBuilder.putFlushId(flushId);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder putLocation(TabletMetadata.Location location) {
+ fetched.add(LOCATION);
+ internalBuilder.putLocation(location);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteLocation(TabletMetadata.Location location) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putZooLock(String zookeeperRoot, ServiceLock zooLock) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putDirName(String dirName) {
+ fetched.add(DIR);
+ internalBuilder.putDirName(dirName);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder putWal(LogEntry logEntry) {
+ fetched.add(LOGS);
+ internalBuilder.putWal(logEntry);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteWal(String wal) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteWal(LogEntry logEntry) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putTime(MetadataTime time) {
+ fetched.add(TIME);
+ internalBuilder.putTime(time);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder putBulkFile(ReferencedTabletFile bulkref, long tid) {
+ fetched.add(LOADED);
+ internalBuilder.putBulkFile(bulkref, tid);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteBulkFile(StoredTabletFile bulkref) {
+ throw new UnsupportedOperationException();
+ }
+
- @Override
- public TabletMetadataBuilder putChopped() {
- fetched.add(CHOPPED);
- internalBuilder.putChopped();
- return this;
- }
-
+ @Override
+ public TabletMetadataBuilder putSuspension(TServerInstance tserver, long suspensionTime) {
+ fetched.add(SUSPEND);
+ internalBuilder.putSuspension(tserver, suspensionTime);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteSuspension() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putExternalCompaction(ExternalCompactionId ecid,
+ ExternalCompactionMetadata ecMeta) {
+ fetched.add(ECOMP);
+ internalBuilder.putExternalCompaction(ecid, ecMeta);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteExternalCompaction(ExternalCompactionId ecid) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putCompacted(long fateTxId) {
+ fetched.add(COMPACTED);
+ internalBuilder.putCompacted(fateTxId);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteCompacted(long fateTxId) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putHostingGoal(TabletHostingGoal goal) {
+ fetched.add(HOSTING_GOAL);
+ internalBuilder.putHostingGoal(goal);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder setHostingRequested() {
+ fetched.add(HOSTING_REQUESTED);
+ internalBuilder.setHostingRequested();
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteHostingRequested() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putOperation(TabletOperationId opId) {
+ fetched.add(OPID);
+ internalBuilder.putOperation(opId);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteOperation() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TabletMetadataBuilder putSelectedFiles(SelectedFiles selectedFiles) {
+ fetched.add(SELECTED);
+ internalBuilder.putSelectedFiles(selectedFiles);
+ return this;
+ }
+
+ @Override
+ public TabletMetadataBuilder deleteSelectedFiles() {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * @param extraFetched Anything that was put on the builder will automatically be added to the
+ * fetched set. However, for the case where something was not put and it needs to be
+ * fetched it can be passed here. For example to simulate a tablet w/o a location it, no
+ * location will be put and LOCATION would be passed in via this argument.
+ */
+ public TabletMetadata build(TabletMetadata.ColumnType... extraFetched) {
+ var mutation = internalBuilder.getMutation();
+
+ SortedMap<Key,Value> rowMap = new TreeMap<>();
+ mutation.getUpdates().forEach(cu -> {
+ Key k = new Key(mutation.getRow(), cu.getColumnFamily(), cu.getColumnQualifier(),
+ cu.getTimestamp());
+ Value v = new Value(cu.getValue());
+ rowMap.put(k, v);
+ });
+
+ fetched.addAll(Arrays.asList(extraFetched));
+
+ return TabletMetadata.convertRow(rowMap.entrySet().iterator(), fetched, true, false);
+ }
+
+}
diff --cc core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMutatorBase.java
index 5969632555,e647d34321..84b10fe3b5
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMutatorBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMutatorBase.java
@@@ -29,9 -27,12 +29,8 @@@ import org.apache.accumulo.core.metadat
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.SuspendingTServer;
import org.apache.accumulo.core.metadata.TServerInstance;
-import org.apache.accumulo.core.metadata.schema.Ample;
-import org.apache.accumulo.core.metadata.schema.Ample.TabletMutator;
-import org.apache.accumulo.core.metadata.schema.DataFileValue;
-import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
-import org.apache.accumulo.core.metadata.schema.ExternalCompactionMetadata;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.BulkFileColumnFamily;
- import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CompactedColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CurrentLocationColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ExternalCompactionColumnFamily;
@@@ -88,39 -80,39 +87,39 @@@ public abstract class TabletMutatorBase
}
@Override
- public Ample.TabletMutator putFile(ReferencedTabletFile path, DataFileValue dfv) {
+ public T putFile(ReferencedTabletFile path, DataFileValue dfv) {
Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
- mutation.put(DataFileColumnFamily.NAME, path.getMetaInsertText(), new Value(dfv.encode()));
+ mutation.put(DataFileColumnFamily.NAME, path.insert().getMetadataText(),
+ new Value(dfv.encode()));
- return this;
+ return getThis();
}
@Override
- public Ample.TabletMutator putFile(StoredTabletFile path, DataFileValue dfv) {
+ public T putFile(StoredTabletFile path, DataFileValue dfv) {
Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
- mutation.put(DataFileColumnFamily.NAME, path.getMetaUpdateDeleteText(),
- new Value(dfv.encode()));
+ mutation.put(DataFileColumnFamily.NAME, path.getMetadataText(), new Value(dfv.encode()));
- return this;
+ return getThis();
}
@Override
- public Ample.TabletMutator deleteFile(StoredTabletFile path) {
+ public T deleteFile(StoredTabletFile path) {
Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
- mutation.putDelete(DataFileColumnFamily.NAME, path.getMetaUpdateDeleteText());
+ mutation.putDelete(DataFileColumnFamily.NAME, path.getMetadataText());
- return this;
+ return getThis();
}
@Override
- public Ample.TabletMutator putScan(StoredTabletFile path) {
+ public T putScan(StoredTabletFile path) {
Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
- mutation.put(ScanFileColumnFamily.NAME, path.getMetaUpdateDeleteText(), new Value());
+ mutation.put(ScanFileColumnFamily.NAME, path.getMetadataText(), new Value());
- return this;
+ return getThis();
}
@Override
- public Ample.TabletMutator deleteScan(StoredTabletFile path) {
+ public T deleteScan(StoredTabletFile path) {
Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
- mutation.putDelete(ScanFileColumnFamily.NAME, path.getMetaUpdateDeleteText());
+ mutation.putDelete(ScanFileColumnFamily.NAME, path.getMetadataText());
- return this;
+ return getThis();
}
@Override
@@@ -202,46 -194,22 +201,39 @@@
}
@Override
- public Ample.TabletMutator putBulkFile(ReferencedTabletFile bulkref, long tid) {
+ public T putBulkFile(ReferencedTabletFile bulkref, long tid) {
Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
- mutation.put(BulkFileColumnFamily.NAME, bulkref.getMetaInsertText(),
+ mutation.put(BulkFileColumnFamily.NAME, bulkref.insert().getMetadataText(),
new Value(FateTxId.formatTid(tid)));
- return this;
+ return getThis();
}
@Override
- public Ample.TabletMutator deleteBulkFile(StoredTabletFile bulkref) {
+ public T deleteBulkFile(StoredTabletFile bulkref) {
Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
- mutation.putDelete(BulkFileColumnFamily.NAME, bulkref.getMetaUpdateDeleteText());
+ mutation.putDelete(BulkFileColumnFamily.NAME, bulkref.getMetadataText());
- return this;
+ return getThis();
+ }
+
+ @Override
+ public T putSelectedFiles(SelectedFiles selectedFiles) {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ mutation.put(ServerColumnFamily.SELECTED_COLUMN.getColumnFamily(),
+ ServerColumnFamily.SELECTED_COLUMN.getColumnQualifier(),
+ new Value(selectedFiles.getMetadataValue()));
+ return getThis();
}
@Override
- public Ample.TabletMutator putSuspension(TServerInstance tServer, long suspensionTime) {
+ public T deleteSelectedFiles() {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ mutation.putDelete(ServerColumnFamily.SELECTED_COLUMN.getColumnFamily(),
+ ServerColumnFamily.SELECTED_COLUMN.getColumnQualifier());
+ return getThis();
+ }
+
- @Override
- public T putChopped() {
- Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
- ChoppedColumnFamily.CHOPPED_COLUMN.put(mutation, new Value("chopped"));
- return getThis();
- }
-
+ @Override
+ public T putSuspension(TServerInstance tServer, long suspensionTime) {
Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
mutation.put(SuspendLocationColumn.SUSPEND_COLUMN.getColumnFamily(),
SuspendLocationColumn.SUSPEND_COLUMN.getColumnQualifier(),
diff --cc core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java
index ee517f27dc,f431cb052c..a3db02c2b2
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java
@@@ -71,9 -69,7 +71,8 @@@ import org.apache.accumulo.core.metadat
import org.apache.accumulo.core.metadata.schema.Ample.ReadConsistency;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.BulkFileColumnFamily;
- import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ClonedColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CompactedColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CurrentLocationColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ExternalCompactionColumnFamily;
diff --cc core/src/main/java/org/apache/accumulo/core/spi/compaction/DefaultCompactionPlanner.java
index e3f5f6535a,ad512da818..0f84f8ed05
--- a/core/src/main/java/org/apache/accumulo/core/spi/compaction/DefaultCompactionPlanner.java
+++ b/core/src/main/java/org/apache/accumulo/core/spi/compaction/DefaultCompactionPlanner.java
@@@ -274,93 -217,75 +274,92 @@@ public class DefaultCompactionPlanner i
Set<CompactableFile> filesCopy = new HashSet<>(params.getCandidates());
+ FakeFileGenerator fakeFileGenerator = new FakeFileGenerator();
+
long maxSizeToCompact = getMaxSizeToCompact(params.getKind());
- Collection<CompactableFile> group;
- if (params.getRunningCompactions().isEmpty()) {
- group =
- findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact, maxSizeToCompact);
+ // This set represents future files that will be produced by running compactions. If the optimal
+ // set of files to compact is computed and contains one of these files, then its optimal to wait
+ // for this compaction to finish.
+ Set<CompactableFile> expectedFiles = new HashSet<>();
+ params.getRunningCompactions().stream().filter(job -> job.getKind() == params.getKind())
+ .map(job -> getExpected(job.getFiles(), fakeFileGenerator))
+ .forEach(compactableFile -> Preconditions.checkState(expectedFiles.add(compactableFile)));
+ Preconditions.checkState(Collections.disjoint(expectedFiles, filesCopy));
+ filesCopy.addAll(expectedFiles);
- if (!group.isEmpty() && group.size() < params.getCandidates().size()
- && params.getCandidates().size() <= maxFilesToCompact
- && (params.getKind() == CompactionKind.USER
- || params.getKind() == CompactionKind.SELECTOR)) {
- // USER and SELECTOR compactions must eventually compact all files. When a subset of files
- // that meets the compaction ratio is selected, look ahead and see if the next compaction
- // would also meet the compaction ratio. If not then compact everything to avoid doing
- // more than logarithmic work across multiple comapctions.
-
- filesCopy.removeAll(group);
- filesCopy.add(getExpected(group, 0));
-
- if (findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact,
- maxSizeToCompact).isEmpty()) {
- // The next possible compaction does not meet the compaction ratio, so compact
- // everything.
- group = Set.copyOf(params.getCandidates());
- }
+ List<Collection<CompactableFile>> compactionJobs = new ArrayList<>();
+ while (true) {
+ var filesToCompact =
+ findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact, maxSizeToCompact);
+ if (!Collections.disjoint(filesToCompact, expectedFiles)) {
+ // the optimal set of files to compact includes the output of a running compaction, so lets
+ // wait for that running compaction to finish.
+ break;
}
- } else if (params.getKind() == CompactionKind.SYSTEM) {
- // This code determines if once the files compacting finish would they be included in a
- // compaction with the files smaller than them? If so, then wait for the running compaction
- // to complete.
+ if (filesToCompact.isEmpty()) {
+ break;
+ }
- // The set of files running compactions may produce
- var expectedFiles = getExpected(params.getRunningCompactions());
+ filesCopy.removeAll(filesToCompact);
- if (!Collections.disjoint(filesCopy, expectedFiles)) {
- throw new AssertionError();
- }
+ // A compaction job will be created for these files, so lets add an expected file for that
+ // planned compaction job. Then if future iterations of this loop will include that file then
+ // they will not compact.
+ var expectedFile = getExpected(filesToCompact, fakeFileGenerator);
+ Preconditions.checkState(expectedFiles.add(expectedFile));
+ Preconditions.checkState(filesCopy.add(expectedFile));
- filesCopy.addAll(expectedFiles);
+ compactionJobs.add(filesToCompact);
- group =
- findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact, maxSizeToCompact);
+ if (filesToCompact.size() < maxFilesToCompact) {
+ // Only continue looking for more compaction jobs when a set of files is found equals
+ // maxFilesToCompact in size. When the files found is less than the max size its an
+ // indication that the compaction ratio was no longer met and therefore it would be
+ // suboptimal to look for more jobs because the smallest optimal set was found.
+ break;
+ }
+ }
- if (!Collections.disjoint(group, expectedFiles)) {
- // file produced by running compaction will eventually compact with existing files, so
- // wait.
- group = Set.of();
+ if (compactionJobs.size() == 1
+ && (params.getKind() == CompactionKind.USER || params.getKind() == CompactionKind.SELECTOR)
+ && compactionJobs.get(0).size() < params.getCandidates().size()
+ && compactionJobs.get(0).size() <= maxFilesToCompact) {
+ // USER and SELECTOR compactions must eventually compact all files. When a subset of files
+ // that meets the compaction ratio is selected, look ahead and see if the next compaction
+ // would also meet the compaction ratio. If not then compact everything to avoid doing
+ // more than logarithmic work across multiple comapctions.
+
+ var group = compactionJobs.get(0);
+ var candidatesCopy = new HashSet<>(params.getCandidates());
+
+ candidatesCopy.removeAll(group);
+ Preconditions.checkState(candidatesCopy.add(getExpected(group, fakeFileGenerator)));
+
+ if (findDataFilesToCompact(candidatesCopy, params.getRatio(), maxFilesToCompact,
+ maxSizeToCompact).isEmpty()) {
+ // The next possible compaction does not meet the compaction ratio, so compact
+ // everything.
+ compactionJobs.set(0, Set.copyOf(params.getCandidates()));
}
- } else {
- group = Set.of();
}
- if (group.isEmpty()
+ if (compactionJobs.isEmpty()
- && (params.getKind() == CompactionKind.USER || params.getKind() == CompactionKind.SELECTOR
- || params.getKind() == CompactionKind.CHOP)
+ && (params.getKind() == CompactionKind.USER || params.getKind() == CompactionKind.SELECTOR)
&& params.getRunningCompactions().stream()
.noneMatch(job -> job.getKind() == params.getKind())) {
- group = findMaximalRequiredSetToCompact(params.getCandidates(), maxFilesToCompact);
+ // These kinds of compaction require files to compact even if none of the files meet the
+ // compaction ratio. No files were found using the compaction ratio and no compactions are
+ // running, so force a compaction.
+ compactionJobs = findMaximalRequiredSetToCompact(params.getCandidates(), maxFilesToCompact);
}
- if (group.isEmpty()) {
- return params.createPlanBuilder().build();
- } else {
- // determine which executor to use based on the size of the files
- var ceid = getExecutor(group);
-
- return params.createPlanBuilder().addJob(createPriority(params, group), ceid, group).build();
- }
+ var builder = params.createPlanBuilder();
+ compactionJobs.forEach(jobFiles -> builder.addJob(createPriority(params, jobFiles),
+ getExecutor(jobFiles), jobFiles));
+ return builder.build();
}
private static short createPriority(PlanningParameters params,
diff --cc core/src/test/java/org/apache/accumulo/core/metadata/schema/SelectedFilesTest.java
index 555f3b2239,0000000000..86483f2c91
mode 100644,000000..100644
--- a/core/src/test/java/org/apache/accumulo/core/metadata/schema/SelectedFilesTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/metadata/schema/SelectedFilesTest.java
@@@ -1,207 -1,0 +1,214 @@@
+/*
+ * 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.core.metadata.schema;
+
+import static org.apache.accumulo.core.util.LazySingletons.RANDOM;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.apache.accumulo.core.fate.FateTxId;
++import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.StoredTabletFile;
++import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class SelectedFilesTest {
+
+ final String filePathPrefix = "hdfs://namenode:9020/accumulo/tables/1/default_tablet/";
+
+ /**
+ * Make sure a SelectedFiles object is equal to another SelectedFiles object that was created
+ * using its metadata json
+ */
+ @Test
+ public void testSerializationDeserialization() {
+ Set<StoredTabletFile> files = getStoredTabletFiles(2);
+
+ SelectedFiles original = new SelectedFiles(files, true, 12345L);
+
+ String json = original.getMetadataValue();
+ SelectedFiles deserialized = SelectedFiles.from(json);
+
+ assertEquals(original, deserialized);
+ assertEquals(json, deserialized.getMetadataValue());
+ }
+
+ /**
+ * Ensure two SelectedFiles objects created with the same parameters are equal
+ */
+ @Test
+ public void testEqualSerialization() {
+ Set<StoredTabletFile> files = getStoredTabletFiles(16);
+
+ SelectedFiles sf1 = new SelectedFiles(files, true, 12345L);
+ SelectedFiles sf2 = new SelectedFiles(files, true, 12345L);
+
+ assertEquals(sf1.getMetadataValue(), sf2.getMetadataValue());
+ assertEquals(sf1, sf2);
+ }
+
+ /**
+ * Ensure that SelectedFiles objects created with file arrays of differing order yet equal entries
+ * are correctly serialized
+ */
+ @Test
+ public void testDifferentFilesOrdering() {
+ Set<StoredTabletFile> files = getStoredTabletFiles(16);
+ SortedSet<StoredTabletFile> sortedFiles = new TreeSet<>(files);
+
+ assertEquals(files, sortedFiles, "Entries in test file sets should be the same");
+ assertNotEquals(files.toString(), sortedFiles.toString(),
+ "Order of files set should differ for this test case");
+
+ SelectedFiles sf1 = new SelectedFiles(files, false, 654123L);
+ SelectedFiles sf2 = new SelectedFiles(sortedFiles, false, 654123L);
+
+ assertEquals(sf1.getMetadataValue(), sf2.getMetadataValue());
+ assertEquals(sf1, sf2);
+ }
+
+ /**
+ * Ensure that two SelectedFiles objects, one created with a subset of files present in the other,
+ * are not equal
+ */
+ @Test
+ public void testJsonSuperSetSubset() {
+ Set<StoredTabletFile> filesSuperSet = getStoredTabletFiles(3);
+ Set<StoredTabletFile> filesSubSet = new HashSet<>(filesSuperSet);
+ // Remove an element to create a subset
+ filesSubSet.remove(filesSubSet.iterator().next());
+
+ SelectedFiles superSetSelectedFiles = new SelectedFiles(filesSuperSet, true, 123456L);
+ SelectedFiles subSetSelectedFiles = new SelectedFiles(filesSubSet, true, 123456L);
+
+ String superSetJson = superSetSelectedFiles.getMetadataValue();
+ String subSetJson = subSetSelectedFiles.getMetadataValue();
+
+ assertNotEquals(superSetJson, subSetJson, "The metadata jsons should have differing file sets");
+
+ // Deserialize the JSON strings back to SelectedFiles and check for equality with the original
+ // objects
+ SelectedFiles superSetDeserialized = SelectedFiles.from(superSetJson);
+ SelectedFiles subSetDeserialized = SelectedFiles.from(subSetJson);
+
+ assertEquals(superSetSelectedFiles, superSetDeserialized);
+ assertEquals(subSetSelectedFiles, subSetDeserialized);
+
+ assertNotEquals(superSetDeserialized, subSetDeserialized,
+ "The objects should have differing file sets");
+ }
+
+ private static Stream<Arguments> provideTestJsons() {
+ return Stream.of(Arguments.of("123456", true, 12), Arguments.of("123456", false, 12),
+ Arguments.of("123456", false, 23), Arguments.of("654321", false, 23),
+ Arguments.of("AE56E", false, 23));
+ }
+
+ /**
+ * Ensure that SelectedFiles objects that are created using {@link SelectedFiles#from(String)} are
+ * created correctly
+ */
+ @ParameterizedTest
+ @MethodSource("provideTestJsons")
+ public void testJsonStrings(String txid, boolean selAll, int numPaths) {
+ List<String> paths = getFilePaths(numPaths);
+
+ // should be resilient to unordered file arrays
+ Collections.shuffle(paths, RANDOM.get());
+
+ // construct a json from the given parameters
+ String json = getJson(txid, selAll, paths);
+
++ System.out.println(json);
++
+ // create a SelectedFiles object using the json
+ SelectedFiles selectedFiles = SelectedFiles.from(json);
+
+ // ensure all parts of the SelectedFiles object are correct
+ assertEquals(Long.parseLong(txid, 16), selectedFiles.getFateTxId());
+ assertEquals(selAll, selectedFiles.initiallySelectedAll());
+ Set<StoredTabletFile> expectedStoredTabletFiles = filePathsToStoredTabletFiles(paths);
+ assertEquals(expectedStoredTabletFiles, selectedFiles.getFiles());
+
+ Collections.sort(paths);
+ String jsonWithSortedFiles = getJson(txid, selAll, paths);
+ assertEquals(jsonWithSortedFiles, selectedFiles.getMetadataValue());
+ }
+
+ /**
+ * Creates a json suitable to create a SelectedFiles object from. The given parameters will be
+ * inserted into the returned json String. Example of returned json String:
+ *
+ * <pre>
+ * {
+ * "txid": "FATE[123456]",
+ * "selAll": true,
+ * "files": ["/path/to/file1.rf", "/path/to/file2.rf"]
+ * }
+ * </pre>
+ */
+ private static String getJson(String txid, boolean selAll, List<String> paths) {
+ String filesJsonArray =
- paths.stream().map(path -> "'" + path + "'").collect(Collectors.joining(","));
++ paths.stream().map(path -> new ReferencedTabletFile(new Path(path)).insert().getMetadata())
++ .map(path -> path.replace("\"", "\\\"")).map(path -> "'" + path + "'")
++ .collect(Collectors.joining(","));
+ return ("{'txid':'" + FateTxId.formatTid(Long.parseLong(txid, 16)) + "','selAll':" + selAll
+ + ",'files':[" + filesJsonArray + "]}").replace('\'', '\"');
+ }
+
+ /**
+ * Generate the given number of rfile path strings
+ */
+ private List<String> getFilePaths(int numOfFiles) {
+ List<String> paths = new ArrayList<>(numOfFiles);
+ for (int i = 0; i < numOfFiles; i++) {
+ paths.add(filePathPrefix + "F0000" + i + ".rf");
+ }
+ return paths;
+ }
+
+ /**
+ * Convert a list of file paths to StoredTabletFile objects
+ */
+ private Set<StoredTabletFile> filePathsToStoredTabletFiles(List<String> filePaths) {
- return filePaths.stream().map(StoredTabletFile::new).collect(Collectors.toSet());
++ return filePaths.stream().map(Path::new).map(ReferencedTabletFile::new)
++ .map(ReferencedTabletFile::insert).collect(Collectors.toSet());
+ }
+
+ /**
+ * Generate the given number of StoredTabletFile objects
+ */
+ private Set<StoredTabletFile> getStoredTabletFiles(int numOfFiles) {
+ return filePathsToStoredTabletFiles(getFilePaths(numOfFiles));
+ }
+
+}
diff --cc core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
index 3e53f4d12e,3b856726be..01f3e9b7b5
--- a/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
@@@ -19,7 -19,7 +19,8 @@@
package org.apache.accumulo.core.metadata.schema;
import static java.util.stream.Collectors.toSet;
+import static org.apache.accumulo.core.fate.FateTxId.formatTid;
+ import static org.apache.accumulo.core.metadata.StoredTabletFile.serialize;
import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.COMPACT_COLUMN;
import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN;
import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.FLUSH_COLUMN;
@@@ -92,10 -80,10 +93,10 @@@ public class TabletMetadataTest
FLUSH_COLUMN.put(mutation, new Value("6"));
TIME_COLUMN.put(mutation, new Value("M123456789"));
- String bf1 = "hdfs://nn1/acc/tables/1/t-0001/bf1";
- String bf2 = "hdfs://nn1/acc/tables/1/t-0001/bf2";
+ String bf1 = serialize("hdfs://nn1/acc/tables/1/t-0001/bf1");
+ String bf2 = serialize("hdfs://nn1/acc/tables/1/t-0001/bf2");
- mutation.at().family(BulkFileColumnFamily.NAME).qualifier(bf1).put(FateTxId.formatTid(56));
- mutation.at().family(BulkFileColumnFamily.NAME).qualifier(bf2).put(FateTxId.formatTid(59));
+ mutation.at().family(BulkFileColumnFamily.NAME).qualifier(bf1).put(formatTid(56));
+ mutation.at().family(BulkFileColumnFamily.NAME).qualifier(bf2).put(formatTid(59));
mutation.at().family(ClonedColumnFamily.NAME).qualifier("").put("OK");
@@@ -310,94 -265,4 +309,98 @@@
});
return rowMap;
}
+
+ @Test
+ public void testBuilder() {
+ TServerInstance ser1 = new TServerInstance(HostAndPort.fromParts("server1", 8555), "s001");
+
+ KeyExtent extent = new KeyExtent(TableId.of("5"), new Text("df"), new Text("da"));
+
- StoredTabletFile sf1 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-0001/sf1.rf");
++ StoredTabletFile sf1 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/sf1.rf")).insert();
+ DataFileValue dfv1 = new DataFileValue(89, 67);
+
- StoredTabletFile sf2 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-0001/sf2.rf");
++ StoredTabletFile sf2 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/sf2.rf")).insert();
+ DataFileValue dfv2 = new DataFileValue(890, 670);
+
+ ReferencedTabletFile rf1 =
+ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/imp1.rf"));
+ ReferencedTabletFile rf2 =
+ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/imp2.rf"));
+
- StoredTabletFile sf3 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-0001/sf3.rf");
- StoredTabletFile sf4 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-0001/sf4.rf");
++ StoredTabletFile sf3 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/sf3.rf")).insert();
++ StoredTabletFile sf4 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/sf4.rf")).insert();
+
+ TabletMetadata tm = TabletMetadata.builder(extent).putHostingGoal(TabletHostingGoal.NEVER)
+ .putLocation(Location.future(ser1)).putFile(sf1, dfv1).putFile(sf2, dfv2)
+ .putCompactionId(23).putBulkFile(rf1, 25).putBulkFile(rf2, 35).putFlushId(27)
+ .putDirName("dir1").putScan(sf3).putScan(sf4).putCompacted(17).putCompacted(23)
+ .build(ECOMP, HOSTING_REQUESTED);
+
+ assertEquals(extent, tm.getExtent());
+ assertEquals(TabletHostingGoal.NEVER, tm.getHostingGoal());
+ assertEquals(Location.future(ser1), tm.getLocation());
+ assertEquals(23L, tm.getCompactId().orElse(-1));
+ assertEquals(27L, tm.getFlushId().orElse(-1));
+ assertEquals(Map.of(sf1, dfv1, sf2, dfv2), tm.getFilesMap());
+ assertEquals(Map.of(rf1.insert(), 25L, rf2.insert(), 35L), tm.getLoaded());
+ assertEquals("dir1", tm.getDirName());
+ assertEquals(Set.of(sf3, sf4), Set.copyOf(tm.getScans()));
+ assertEquals(Set.of(), tm.getExternalCompactions().keySet());
+ assertEquals(Set.of(17L, 23L), tm.getCompacted());
+ assertFalse(tm.getHostingRequested());
+ assertThrows(IllegalStateException.class, tm::getOperationId);
+ assertThrows(IllegalStateException.class, tm::getSuspend);
+ assertThrows(IllegalStateException.class, tm::getTime);
+
+ TabletOperationId opid1 = TabletOperationId.from(TabletOperationType.SPLITTING, 55);
+ TabletMetadata tm2 = TabletMetadata.builder(extent).putOperation(opid1).build(LOCATION);
+
+ assertEquals(extent, tm2.getExtent());
+ assertEquals(opid1, tm2.getOperationId());
+ assertNull(tm2.getLocation());
+ assertThrows(IllegalStateException.class, tm2::getFiles);
+ assertThrows(IllegalStateException.class, tm2::getHostingGoal);
+ assertThrows(IllegalStateException.class, tm2::getCompactId);
+ assertThrows(IllegalStateException.class, tm2::getFlushId);
+ assertThrows(IllegalStateException.class, tm2::getFiles);
+ assertThrows(IllegalStateException.class, tm2::getLogs);
+ assertThrows(IllegalStateException.class, tm2::getLoaded);
+ assertThrows(IllegalStateException.class, tm2::getDirName);
+ assertThrows(IllegalStateException.class, tm2::getScans);
+ assertThrows(IllegalStateException.class, tm2::getExternalCompactions);
+ assertThrows(IllegalStateException.class, tm2::getHostingRequested);
+ assertThrows(IllegalStateException.class, tm2::getSelectedFiles);
+ assertThrows(IllegalStateException.class, tm2::getCompacted);
+
+ var ecid1 = ExternalCompactionId.generate(UUID.randomUUID());
+ ExternalCompactionMetadata ecm = new ExternalCompactionMetadata(Set.of(sf1, sf2), rf1, "cid1",
+ CompactionKind.USER, (short) 3, CompactionExecutorIdImpl.externalId("Q1"), true, 99L);
+
+ LogEntry le1 = new LogEntry(extent, 55, "lf1");
+ LogEntry le2 = new LogEntry(extent, 57, "lf2");
+
+ SelectedFiles selFiles = new SelectedFiles(Set.of(sf1, sf4), false, 159L);
+
+ TabletMetadata tm3 = TabletMetadata.builder(extent).putExternalCompaction(ecid1, ecm)
+ .putSuspension(ser1, 45L).putTime(new MetadataTime(479, TimeType.LOGICAL)).putWal(le1)
+ .putWal(le2).setHostingRequested().putSelectedFiles(selFiles).build();
+
+ assertEquals(Set.of(ecid1), tm3.getExternalCompactions().keySet());
+ assertEquals(Set.of(sf1, sf2), tm3.getExternalCompactions().get(ecid1).getJobFiles());
+ assertEquals(ser1.getHostAndPort(), tm3.getSuspend().server);
+ assertEquals(45L, tm3.getSuspend().suspensionTime);
+ assertEquals(new MetadataTime(479, TimeType.LOGICAL), tm3.getTime());
+ assertTrue(tm3.getHostingRequested());
+ assertEquals(List.of(le1, le2).stream().map(LogEntry::toString).collect(toSet()),
+ tm3.getLogs().stream().map(LogEntry::toString).collect(toSet()));
+ assertEquals(Set.of(sf1, sf4), tm3.getSelectedFiles().getFiles());
+ assertEquals(159L, tm3.getSelectedFiles().getFateTxId());
+ assertFalse(tm3.getSelectedFiles().initiallySelectedAll());
+ assertEquals(selFiles.getMetadataValue(), tm3.getSelectedFiles().getMetadataValue());
+ }
+
}
diff --cc server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
index 688a9680e1,97dd31f203..db71c37730
--- a/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
@@@ -40,9 -40,7 +40,8 @@@ import org.apache.accumulo.core.metadat
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.BulkFileColumnFamily;
- import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ClonedColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CompactedColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CurrentLocationColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ExternalCompactionColumnFamily;
@@@ -101,10 -95,8 +100,9 @@@ public class MetadataConstraints implem
CurrentLocationColumnFamily.NAME,
LastLocationColumnFamily.NAME,
FutureLocationColumnFamily.NAME,
- ChoppedColumnFamily.NAME,
ClonedColumnFamily.NAME,
- ExternalCompactionColumnFamily.NAME);
+ ExternalCompactionColumnFamily.NAME,
+ CompactedColumnFamily.NAME);
// @formatter:on
private static boolean isValidColumn(ColumnUpdate cu) {
@@@ -240,32 -229,19 +238,41 @@@
} else if (columnFamily.equals(ScanFileColumnFamily.NAME)) {
violations = validateDataFileMetadata(violations,
new String(columnUpdate.getColumnQualifier(), UTF_8));
+ } else if (HostingColumnFamily.GOAL_COLUMN.equals(columnFamily, columnQualifier)) {
+ try {
+ TabletHostingGoalUtil.fromValue(new Value(columnUpdate.getValue()));
+ } catch (IllegalArgumentException e) {
+ violations = addViolation(violations, 10);
+ }
+ } else if (ServerColumnFamily.OPID_COLUMN.equals(columnFamily, columnQualifier)) {
+ try {
+ TabletOperationId.validate(new String(columnUpdate.getValue(), UTF_8));
+ } catch (IllegalArgumentException e) {
+ violations = addViolation(violations, 9);
+ }
+ } else if (ServerColumnFamily.SELECTED_COLUMN.equals(columnFamily, columnQualifier)) {
+ try {
+ SelectedFiles.from(new String(columnUpdate.getValue(), UTF_8));
+ } catch (RuntimeException e) {
+ violations = addViolation(violations, 11);
+ }
+ } else if (CompactedColumnFamily.NAME.equals(columnFamily)) {
+ if (!FateTxId.isFormatedTid(columnQualifier.toString())) {
+ violations = addViolation(violations, 13);
+ }
} else if (columnFamily.equals(BulkFileColumnFamily.NAME)) {
if (!columnUpdate.isDeleted() && !checkedBulk) {
- violations = validateDataFileMetadata(violations,
- new String(columnUpdate.getColumnQualifier(), UTF_8));
+ /*
+ * This needs to be re-worked after Issue https://github.com/apache/accumulo/issues/3505
+ * is done.
+ *
+ * That issue will reorganizes this class and make things more efficient so we are not
+ * looping over the same mutation more than once like in this case. The below check is
+ * commented out for now because the violation check is already done when creating
+ * StoredTabletFiles so it isn't needed here anymore violations =
+ * validateDataFileMetadata(violations, new String(columnUpdate.getColumnQualifier(),
+ * UTF_8));
+ */
// splits, which also write the time reference, are allowed to write this reference even
// when
@@@ -293,9 -269,20 +300,20 @@@
.equals(CurrentLocationColumnFamily.NAME)) {
isLocationMutation = true;
} else if (new Text(update.getColumnFamily()).equals(DataFileColumnFamily.NAME)) {
- dataFiles.add(new Text(update.getColumnQualifier()));
+ try {
+ // This actually validates for a second time as the loop already validates
+ // if a DataFileColumnFamily, this will likely be fixed as part of
+ // https://github.com/apache/accumulo/issues/3505
+ dataFiles.add(StoredTabletFile.of(new Text(update.getColumnQualifier())));
+ } catch (RuntimeException e) {
- violations = addViolation(violations, 9);
++ violations = addViolation(violations, 12);
+ }
} else if (new Text(update.getColumnFamily()).equals(BulkFileColumnFamily.NAME)) {
- loadedFiles.add(new Text(update.getColumnQualifier()));
+ try {
+ loadedFiles.add(StoredTabletFile.of(new Text(update.getColumnQualifier())));
+ } catch (RuntimeException e) {
- violations = addViolation(violations, 9);
++ violations = addViolation(violations, 12);
+ }
if (!new String(update.getValue(), UTF_8).equals(tidString)) {
otherTidCount++;
diff --cc server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletManagementIterator.java
index 4d12d5112a,0000000000..ea50b01662
mode 100644,000000..100644
--- a/server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletManagementIterator.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletManagementIterator.java
@@@ -1,528 -1,0 +1,526 @@@
+/*
+ * 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.manager.state;
+
+import static org.apache.accumulo.core.util.LazySingletons.GSON;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.SortedMap;
+
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.ScannerBase;
+import org.apache.accumulo.core.client.admin.TabletHostingGoal;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.ConfigurationCopy;
+import org.apache.accumulo.core.conf.ConfigurationTypeHelper;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Key;
+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.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.SkippingIterator;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.user.WholeRowIterator;
+import org.apache.accumulo.core.manager.balancer.TabletServerIdImpl;
+import org.apache.accumulo.core.manager.state.TabletManagement;
+import org.apache.accumulo.core.manager.state.TabletManagement.ManagementAction;
+import org.apache.accumulo.core.manager.thrift.ManagerState;
+import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.core.metadata.TabletState;
+import org.apache.accumulo.core.metadata.schema.DataFileValue;
- import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CurrentLocationColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ExternalCompactionColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.FutureLocationColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.HostingColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LastLocationColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.SuspendLocationColumn;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.spi.balancer.SimpleLoadBalancer;
+import org.apache.accumulo.core.spi.balancer.TabletBalancer;
+import org.apache.accumulo.core.spi.balancer.data.TabletServerId;
+import org.apache.accumulo.core.spi.compaction.CompactionKind;
+import org.apache.accumulo.core.util.AddressUtil;
+import org.apache.accumulo.server.compaction.CompactionJobGenerator;
+import org.apache.accumulo.server.iterators.TabletIteratorEnvironment;
+import org.apache.accumulo.server.manager.balancer.BalancerEnvironmentImpl;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.Sets;
+import com.google.gson.reflect.TypeToken;
+
+/**
+ * Iterator used by the TabletGroupWatcher threads in the Manager. This iterator returns
+ * TabletManagement objects for each Tablet that needs some type of action performed on it by the
+ * Manager.
+ */
+public class TabletManagementIterator extends SkippingIterator {
+ private static final Logger LOG = LoggerFactory.getLogger(TabletManagementIterator.class);
+
+ private static final String SERVERS_OPTION = "servers";
+ private static final String TABLES_OPTION = "tables";
+ private static final String MERGES_OPTION = "merges";
+ private static final String MIGRATIONS_OPTION = "migrations";
+ private static final String MANAGER_STATE_OPTION = "managerState";
+ private static final String SHUTTING_DOWN_OPTION = "shuttingDown";
+ private static final String RESOURCE_GROUPS = "resourceGroups";
+ private static final String TSERVER_GROUP_PREFIX = "serverGroups_";
+ private static final String COMPACTION_HINTS_OPTIONS = "compactionHints";
+ private CompactionJobGenerator compactionGenerator;
+ private TabletBalancer balancer;
+
+ private static void setCurrentServers(final IteratorSetting cfg,
+ final Set<TServerInstance> goodServers) {
+ if (goodServers != null) {
+ List<String> servers = new ArrayList<>();
+ for (TServerInstance server : goodServers) {
+ servers.add(server.getHostPortSession());
+ }
+ cfg.addOption(SERVERS_OPTION, Joiner.on(",").join(servers));
+ }
+ }
+
+ private static void setOnlineTables(final IteratorSetting cfg, final Set<TableId> onlineTables) {
+ if (onlineTables != null) {
+ cfg.addOption(TABLES_OPTION, Joiner.on(",").join(onlineTables));
+ }
+ }
+
+ private static void setMerges(final IteratorSetting cfg, final Collection<MergeInfo> merges) {
+ DataOutputBuffer buffer = new DataOutputBuffer();
+ try {
+ for (MergeInfo info : merges) {
+ KeyExtent extent = info.getExtent();
+ if (extent != null && !info.getState().equals(MergeState.NONE)) {
+ info.write(buffer);
+ }
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ String encoded =
+ Base64.getEncoder().encodeToString(Arrays.copyOf(buffer.getData(), buffer.getLength()));
+ cfg.addOption(MERGES_OPTION, encoded);
+ }
+
+ private static void setMigrations(final IteratorSetting cfg,
+ final Collection<KeyExtent> migrations) {
+ DataOutputBuffer buffer = new DataOutputBuffer();
+ try {
+ for (KeyExtent extent : migrations) {
+ extent.writeTo(buffer);
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ String encoded =
+ Base64.getEncoder().encodeToString(Arrays.copyOf(buffer.getData(), buffer.getLength()));
+ cfg.addOption(MIGRATIONS_OPTION, encoded);
+ }
+
+ private static void setManagerState(final IteratorSetting cfg, final ManagerState state) {
+ cfg.addOption(MANAGER_STATE_OPTION, state.toString());
+ }
+
+ private static void setShuttingDown(final IteratorSetting cfg,
+ final Set<TServerInstance> servers) {
+ if (servers != null) {
+ cfg.addOption(SHUTTING_DOWN_OPTION, Joiner.on(",").join(servers));
+ }
+ }
+
+ private static void setCompactionHints(final IteratorSetting cfg,
+ Map<Long,Map<String,String>> allHints) {
+ cfg.addOption(COMPACTION_HINTS_OPTIONS, GSON.get().toJson(allHints));
+ }
+
+ private static void setTServerResourceGroups(final IteratorSetting cfg,
+ Map<String,Set<TServerInstance>> tServerResourceGroups) {
+ if (tServerResourceGroups == null) {
+ return;
+ }
+ cfg.addOption(RESOURCE_GROUPS, Joiner.on(",").join(tServerResourceGroups.keySet()));
+ for (Entry<String,Set<TServerInstance>> entry : tServerResourceGroups.entrySet()) {
+ cfg.addOption(TSERVER_GROUP_PREFIX + entry.getKey(), Joiner.on(",").join(entry.getValue()));
+ }
+ }
+
+ private static Map<TabletServerId,String> parseTServerResourceGroups(Map<String,String> options) {
+ Map<TabletServerId,String> resourceGroups = new HashMap<>();
+ String groups = options.get(RESOURCE_GROUPS);
+ if (groups != null) {
+ for (String groupName : groups.split(",")) {
+ String groupServers = options.get(TSERVER_GROUP_PREFIX + groupName);
+ if (groupServers != null) {
+ Set<TServerInstance> servers = parseServers(groupServers);
+ servers.forEach(server -> resourceGroups.put(new TabletServerIdImpl(server), groupName));
+ }
+ }
+ }
+ return resourceGroups;
+ }
+
+ private static Set<KeyExtent> parseMigrations(final String migrations) {
+ Set<KeyExtent> result = new HashSet<>();
+ if (migrations != null) {
+ try {
+ DataInputBuffer buffer = new DataInputBuffer();
+ byte[] data = Base64.getDecoder().decode(migrations);
+ buffer.reset(data, data.length);
+ while (buffer.available() > 0) {
+ result.add(KeyExtent.readFrom(buffer));
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+ return result;
+ }
+
+ private static Set<TableId> parseTableIDs(final String tableIDs) {
+ Set<TableId> result = new HashSet<>();
+ if (tableIDs != null) {
+ for (String tableID : tableIDs.split(",")) {
+ result.add(TableId.of(tableID));
+ }
+ }
+ return result;
+ }
+
+ private static Set<TServerInstance> parseServers(final String servers) {
+ Set<TServerInstance> result = new HashSet<>();
+ if (servers != null) {
+ // parse "host:port[INSTANCE]"
+ if (!servers.isEmpty()) {
+ for (String part : servers.split(",")) {
+ String[] parts = part.split("\\[", 2);
+ String hostport = parts[0];
+ String instance = parts[1];
+ if (instance != null && instance.endsWith("]")) {
+ instance = instance.substring(0, instance.length() - 1);
+ }
+ result.add(new TServerInstance(AddressUtil.parseAddress(hostport, false), instance));
+ }
+ }
+ }
+ return result;
+ }
+
+ private static Map<TableId,MergeInfo> parseMerges(final String merges) {
+ Map<TableId,MergeInfo> result = new HashMap<>();
+ if (merges != null) {
+ try {
+ DataInputBuffer buffer = new DataInputBuffer();
+ byte[] data = Base64.getDecoder().decode(merges);
+ buffer.reset(data, data.length);
+ while (buffer.available() > 0) {
+ MergeInfo mergeInfo = new MergeInfo();
+ mergeInfo.readFields(buffer);
+ result.put(mergeInfo.extent.tableId(), mergeInfo);
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+ return result;
+ }
+
+ private static Map<Long,Map<String,String>> parseCompactionHints(String json) {
+ if (json == null) {
+ return Map.of();
+ }
+ Type tt = new TypeToken<Map<Long,Map<String,String>>>() {}.getType();
+ return GSON.get().fromJson(json, tt);
+ }
+
+ private static boolean shouldReturnDueToSplit(final TabletMetadata tm,
+ final long splitThreshold) {
+ final long sumOfFileSizes =
+ tm.getFilesMap().values().stream().mapToLong(DataFileValue::getSize).sum();
+ final boolean shouldSplit = sumOfFileSizes > splitThreshold;
+ LOG.debug("{} should split? sum: {}, threshold: {}, result: {}", tm.getExtent(), sumOfFileSizes,
+ splitThreshold, shouldSplit);
+ return shouldSplit;
+ }
+
+ private boolean shouldReturnDueToLocation(final TabletMetadata tm,
+ final Set<TableId> onlineTables, final Set<TServerInstance> current) {
+
+ if (migrations.contains(tm.getExtent())) {
+ return true;
+ }
+
+ // is the table supposed to be online or offline?
+ final boolean shouldBeOnline =
+ onlineTables.contains(tm.getTableId()) && tm.getOperationId() == null;
+
+ TabletState state = TabletState.compute(tm, current, balancer, tserverResourceGroups);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("{} is {}. Table is {}line. Tablet hosting goal is {}, hostingRequested: {}",
+ tm.getExtent(), state, (shouldBeOnline ? "on" : "off"), tm.getHostingGoal(),
+ tm.getHostingRequested());
+ }
+ switch (state) {
+ case ASSIGNED:
+ // we always want data about assigned tablets
+ return true;
+ case HOSTED:
+ if (!shouldBeOnline || tm.getHostingGoal() == TabletHostingGoal.NEVER
+ || (tm.getHostingGoal() == TabletHostingGoal.ONDEMAND && !tm.getHostingRequested())) {
+ return true;
+ }
+ break;
+ case ASSIGNED_TO_DEAD_SERVER:
+ case NEEDS_REASSIGNMENT:
+ return true;
+ case SUSPENDED:
+ case UNASSIGNED:
+ if (shouldBeOnline && (tm.getHostingGoal() == TabletHostingGoal.ALWAYS
+ || (tm.getHostingGoal() == TabletHostingGoal.ONDEMAND && tm.getHostingRequested()))) {
+ return true;
+ }
+ break;
+ default:
+ throw new AssertionError("Inconceivable! The tablet is an unrecognized state: " + state);
+ }
+ return false;
+ }
+
+ public static void configureScanner(final ScannerBase scanner, final CurrentState state) {
+ // TODO so many columns are being fetch it may not make sense to fetch columns
+ TabletColumnFamily.PREV_ROW_COLUMN.fetch(scanner);
+ ServerColumnFamily.DIRECTORY_COLUMN.fetch(scanner);
+ ServerColumnFamily.SELECTED_COLUMN.fetch(scanner);
+ scanner.fetchColumnFamily(CurrentLocationColumnFamily.NAME);
+ scanner.fetchColumnFamily(FutureLocationColumnFamily.NAME);
+ scanner.fetchColumnFamily(LastLocationColumnFamily.NAME);
+ scanner.fetchColumnFamily(SuspendLocationColumn.SUSPEND_COLUMN.getColumnFamily());
+ scanner.fetchColumnFamily(LogColumnFamily.NAME);
- scanner.fetchColumnFamily(ChoppedColumnFamily.NAME);
+ scanner.fetchColumnFamily(HostingColumnFamily.NAME);
+ scanner.fetchColumnFamily(DataFileColumnFamily.NAME);
+ scanner.fetchColumnFamily(ExternalCompactionColumnFamily.NAME);
+ ServerColumnFamily.OPID_COLUMN.fetch(scanner);
+ scanner.addScanIterator(new IteratorSetting(1000, "wholeRows", WholeRowIterator.class));
+ IteratorSetting tabletChange =
+ new IteratorSetting(1001, "ManagerTabletInfoIterator", TabletManagementIterator.class);
+ if (state != null) {
+ TabletManagementIterator.setCurrentServers(tabletChange, state.onlineTabletServers());
+ TabletManagementIterator.setOnlineTables(tabletChange, state.onlineTables());
+ TabletManagementIterator.setMerges(tabletChange, state.merges());
+ // ELASTICITY_TODO passing the unassignemnt request as part of the migrations is a hack. Was
+ // not sure of the entire unassignment request approach and did not want to push it further
+ // into the code.
+ TabletManagementIterator.setMigrations(tabletChange,
+ Sets.union(state.migrationsSnapshot(), state.getUnassignmentRequest()));
+ TabletManagementIterator.setManagerState(tabletChange, state.getManagerState());
+ TabletManagementIterator.setShuttingDown(tabletChange, state.shutdownServers());
+ TabletManagementIterator.setTServerResourceGroups(tabletChange,
+ state.tServerResourceGroups());
+ setCompactionHints(tabletChange, state.getCompactionHints());
+ }
+ scanner.addScanIterator(tabletChange);
+ }
+
+ public static TabletManagement decode(Entry<Key,Value> e) throws IOException {
+ return new TabletManagement(e.getKey(), e.getValue());
+ }
+
+ private final Set<TServerInstance> current = new HashSet<>();
+ private final Set<TableId> onlineTables = new HashSet<>();
+ private final Map<TabletServerId,String> tserverResourceGroups = new HashMap<>();
+ private final Map<TableId,MergeInfo> merges = new HashMap<>();
+ private final Set<KeyExtent> migrations = new HashSet<>();
+ private ManagerState managerState = ManagerState.NORMAL;
+ private IteratorEnvironment env;
+ private Key topKey = null;
+ private Value topValue = null;
+
+ @Override
+ public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options,
+ IteratorEnvironment env) throws IOException {
+ super.init(source, options, env);
+ this.env = env;
+ current.addAll(parseServers(options.get(SERVERS_OPTION)));
+ onlineTables.addAll(parseTableIDs(options.get(TABLES_OPTION)));
+ tserverResourceGroups.putAll(parseTServerResourceGroups(options));
+ merges.putAll(parseMerges(options.get(MERGES_OPTION)));
+ migrations.addAll(parseMigrations(options.get(MIGRATIONS_OPTION)));
+ String managerStateOptionValue = options.get(MANAGER_STATE_OPTION);
+ try {
+ managerState = ManagerState.valueOf(managerStateOptionValue);
+ } catch (RuntimeException ex) {
+ if (managerStateOptionValue != null) {
+ LOG.error("Unable to decode managerState {}", managerStateOptionValue);
+ }
+ }
+ Set<TServerInstance> shuttingDown = parseServers(options.get(SHUTTING_DOWN_OPTION));
+ if (shuttingDown != null) {
+ current.removeAll(shuttingDown);
+ }
+ compactionGenerator = new CompactionJobGenerator(env.getPluginEnv(),
+ parseCompactionHints(options.get(COMPACTION_HINTS_OPTIONS)));
+ final AccumuloConfiguration conf = new ConfigurationCopy(env.getPluginEnv().getConfiguration());
+ BalancerEnvironmentImpl benv =
+ new BalancerEnvironmentImpl(((TabletIteratorEnvironment) env).getServerContext());
+ balancer = Property.createInstanceFromPropertyName(conf, Property.MANAGER_TABLET_BALANCER,
+ TabletBalancer.class, new SimpleLoadBalancer());
+ balancer.init(benv);
+ LOG.debug("Balancer is set to {}", balancer.getClass().getSimpleName());
+ }
+
+ @Override
+ public Key getTopKey() {
+ return topKey;
+ }
+
+ @Override
+ public Value getTopValue() {
+ return topValue;
+ }
+
+ @Override
+ public boolean hasTop() {
+ return topKey != null && topValue != null;
+ }
+
+ @Override
+ protected void consume() throws IOException {
+ topKey = null;
+ topValue = null;
+
+ final Set<ManagementAction> actions = new HashSet<>();
+ while (getSource().hasTop()) {
+ final Key k = getSource().getTopKey();
+ final Value v = getSource().getTopValue();
+ final SortedMap<Key,Value> decodedRow = WholeRowIterator.decodeRow(k, v);
+ final TabletMetadata tm = TabletMetadata.convertRow(decodedRow.entrySet().iterator(),
+ TabletManagement.CONFIGURED_COLUMNS, false, true);
+
+ actions.clear();
+ if (managerState != ManagerState.NORMAL || current.isEmpty() || onlineTables.isEmpty()) {
+ // when manager is in the process of starting up or shutting down return everything.
+ actions.add(ManagementAction.NEEDS_LOCATION_UPDATE);
+ } else {
+ LOG.trace("Evaluating extent: {}", tm);
+ computeTabletManagementActions(tm, actions);
+ }
+
+ if (!actions.isEmpty()) {
+ // If we simply returned here, then the client would get the encoded K,V
+ // from the WholeRowIterator. However, it would not know the reason(s) why
+ // it was returned. Insert a K,V pair to represent the reasons. The client
+ // can pull this K,V pair from the results by looking at the colf.
+ TabletManagement.addActions(decodedRow, actions);
+ topKey = decodedRow.firstKey();
+ topValue = WholeRowIterator.encodeRow(new ArrayList<>(decodedRow.keySet()),
+ new ArrayList<>(decodedRow.values()));
+ LOG.trace("Returning extent {} with reasons: {}", tm.getExtent(), actions);
+ return;
+ }
+
+ LOG.trace("No reason to return extent {}, continuing", tm.getExtent());
+ getSource().next();
+ }
+ }
+
+ /**
+ * Evaluates whether or not this Tablet should be returned so that it can be acted upon by the
+ * Manager
+ */
+ private void computeTabletManagementActions(final TabletMetadata tm,
+ final Set<ManagementAction> reasonsToReturnThisTablet) {
+
+ if (tm.isFutureAndCurrentLocationSet() || tm.isOperationIdAndCurrentLocationSet()) {
+ // no need to check everything, we are in a known state where we want to return everything.
+ reasonsToReturnThisTablet.add(ManagementAction.BAD_STATE);
+ return;
+ }
+
+ // we always want data about merges
+ final MergeInfo merge = merges.get(tm.getTableId());
+ if (merge != null) {
+ // could make this smarter by only returning if the tablet is involved in the merge
+ reasonsToReturnThisTablet.add(ManagementAction.IS_MERGING);
+ }
+
+ if (shouldReturnDueToLocation(tm, onlineTables, current)) {
+ reasonsToReturnThisTablet.add(ManagementAction.NEEDS_LOCATION_UPDATE);
+ }
+
+ if (tm.getOperationId() == null) {
+ try {
+ final long splitThreshold =
+ ConfigurationTypeHelper.getFixedMemoryAsBytes(this.env.getPluginEnv()
+ .getConfiguration(tm.getTableId()).get(Property.TABLE_SPLIT_THRESHOLD.getKey()));
+ if (shouldReturnDueToSplit(tm, splitThreshold)) {
+ reasonsToReturnThisTablet.add(ManagementAction.NEEDS_SPLITTING);
+ }
+ // important to call this since reasonsToReturnThisTablet is passed to it
+ if (!compactionGenerator
+ .generateJobs(tm, determineCompactionKinds(reasonsToReturnThisTablet)).isEmpty()) {
+ reasonsToReturnThisTablet.add(ManagementAction.NEEDS_COMPACTING);
+ }
+ } catch (NullPointerException e) {
+ LOG.info(
+ "Unable to determine if tablet {} should split or compact, maybe table was deleted?",
+ tm.getExtent());
+ }
+ }
+ }
+
+ private static final Set<CompactionKind> ALL_COMPACTION_KINDS =
+ Collections.unmodifiableSet(EnumSet.allOf(CompactionKind.class));
+ private static final Set<CompactionKind> SPLIT_COMPACTION_KINDS;
+
+ static {
+ var tmp = EnumSet.allOf(CompactionKind.class);
+ tmp.remove(CompactionKind.SYSTEM);
+ SPLIT_COMPACTION_KINDS = Collections.unmodifiableSet(tmp);
+ }
+
+ public static Set<CompactionKind>
+ determineCompactionKinds(Set<ManagementAction> reasonsToReturnThisTablet) {
+ if (reasonsToReturnThisTablet.contains(ManagementAction.NEEDS_SPLITTING)) {
+ return SPLIT_COMPACTION_KINDS;
+ } else {
+ return ALL_COMPACTION_KINDS;
+ }
+ }
+}
diff --cc server/base/src/main/java/org/apache/accumulo/server/metadata/ConditionalTabletMutatorImpl.java
index b5fc0c419c,0000000000..7cbe7b298c
mode 100644,000000..100644
--- a/server/base/src/main/java/org/apache/accumulo/server/metadata/ConditionalTabletMutatorImpl.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/metadata/ConditionalTabletMutatorImpl.java
@@@ -1,250 -1,0 +1,249 @@@
+/*
+ * 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.schema.MetadataSchema.TabletsSection.HostingColumnFamily.GOAL_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.COMPACT_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.OPID_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.SELECTED_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily.encodePrevEndRow;
+
+import java.util.Objects;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.admin.TabletHostingGoal;
+import org.apache.accumulo.core.clientImpl.TabletHostingGoalUtil;
+import org.apache.accumulo.core.data.Condition;
+import org.apache.accumulo.core.data.ConditionalMutation;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.fate.FateTxId;
+import org.apache.accumulo.core.iterators.SortedFilesIterator;
+import org.apache.accumulo.core.metadata.schema.Ample;
+import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.BulkFileColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CompactedColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ExternalCompactionColumnFamily;
+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.metadata.schema.TabletMutatorBase;
+import org.apache.accumulo.core.metadata.schema.TabletOperationId;
- import org.apache.accumulo.core.util.TextUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.metadata.iterators.LocationExistsIterator;
+import org.apache.accumulo.server.metadata.iterators.PresentIterator;
+import org.apache.accumulo.server.metadata.iterators.SetEqualityIterator;
+import org.apache.accumulo.server.metadata.iterators.TabletExistsIterator;
+import org.apache.hadoop.io.Text;
+
+import com.google.common.base.Preconditions;
+
+public class ConditionalTabletMutatorImpl extends TabletMutatorBase<Ample.ConditionalTabletMutator>
+ implements Ample.ConditionalTabletMutator, Ample.OperationRequirements {
+
+ public static final int INITIAL_ITERATOR_PRIO = 1000000;
+
+ private final ConditionalMutation mutation;
+ private final Consumer<ConditionalMutation> mutationConsumer;
+ private final Ample.ConditionalTabletsMutator parent;
+
+ private final BiConsumer<KeyExtent,Ample.RejectionHandler> rejectionHandlerConsumer;
+
+ private final KeyExtent extent;
+
+ private boolean sawOperationRequirement = false;
+
+ protected ConditionalTabletMutatorImpl(Ample.ConditionalTabletsMutator parent,
+ ServerContext context, KeyExtent extent, Consumer<ConditionalMutation> mutationConsumer,
+ BiConsumer<KeyExtent,Ample.RejectionHandler> rejectionHandlerConsumer) {
+ super(new ConditionalMutation(extent.toMetaRow()));
+ this.mutation = (ConditionalMutation) super.mutation;
+ this.mutationConsumer = mutationConsumer;
+ this.parent = parent;
+ this.rejectionHandlerConsumer = rejectionHandlerConsumer;
+ this.extent = extent;
+ }
+
+ @Override
+ public Ample.ConditionalTabletMutator requireAbsentLocation() {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ IteratorSetting is = new IteratorSetting(INITIAL_ITERATOR_PRIO, LocationExistsIterator.class);
+ Condition c = new Condition("", "").setIterators(is);
+ mutation.addCondition(c);
+ return this;
+ }
+
+ @Override
+ public Ample.ConditionalTabletMutator requireLocation(Location location) {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ Preconditions.checkArgument(location.getType() == TabletMetadata.LocationType.FUTURE
+ || location.getType() == TabletMetadata.LocationType.CURRENT);
+ Condition c = new Condition(getLocationFamily(location.getType()), location.getSession())
+ .setValue(location.getHostPort());
+ mutation.addCondition(c);
+ return this;
+ }
+
+ @Override
+ public Ample.ConditionalTabletMutator requirePrevEndRow(Text per) {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ Condition c =
+ new Condition(PREV_ROW_COLUMN.getColumnFamily(), PREV_ROW_COLUMN.getColumnQualifier())
+ .setValue(encodePrevEndRow(per).get());
+ mutation.addCondition(c);
+ return this;
+ }
+
+ @Override
+ public Ample.ConditionalTabletMutator requireHostingGoal(TabletHostingGoal goal) {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ Condition c = new Condition(GOAL_COLUMN.getColumnFamily(), GOAL_COLUMN.getColumnQualifier())
+ .setValue(TabletHostingGoalUtil.toValue(goal).get());
+ mutation.addCondition(c);
+ return this;
+ }
+
+ @Override
+ public Ample.ConditionalTabletMutator requireCompaction(ExternalCompactionId ecid) {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ IteratorSetting is = new IteratorSetting(INITIAL_ITERATOR_PRIO, PresentIterator.class);
+ Condition c = new Condition(ExternalCompactionColumnFamily.STR_NAME, ecid.canonical())
+ .setValue(PresentIterator.VALUE).setIterators(is);
+ mutation.addCondition(c);
+ return this;
+ }
+
+ @Override
+ public Ample.ConditionalTabletMutator requireAbsentTablet() {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ IteratorSetting is = new IteratorSetting(INITIAL_ITERATOR_PRIO, TabletExistsIterator.class);
+ Condition c = new Condition("", "").setIterators(is);
+ mutation.addCondition(c);
+ sawOperationRequirement = true;
+ return this;
+ }
+
+ @Override
+ public Ample.ConditionalTabletMutator requireAbsentOperation() {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ Condition c = new Condition(OPID_COLUMN.getColumnFamily(), OPID_COLUMN.getColumnQualifier());
+ mutation.addCondition(c);
+ sawOperationRequirement = true;
+ return this;
+ }
+
+ @Override
+ public Ample.ConditionalTabletMutator requireOperation(TabletOperationId opid) {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ Condition c = new Condition(OPID_COLUMN.getColumnFamily(), OPID_COLUMN.getColumnQualifier())
+ .setValue(opid.canonical());
+ mutation.addCondition(c);
+ sawOperationRequirement = true;
+ return this;
+ }
+
+ private void requireSameSingle(TabletMetadata tabletMetadata, ColumnType type) {
+ switch (type) {
+ case PREV_ROW:
+ requirePrevEndRow(tabletMetadata.getPrevEndRow());
+ break;
+ case COMPACT_ID: {
+ Condition c =
+ new Condition(COMPACT_COLUMN.getColumnFamily(), COMPACT_COLUMN.getColumnQualifier());
+ if (tabletMetadata.getCompactId().isPresent()) {
+ c = c.setValue(Long.toString(tabletMetadata.getCompactId().getAsLong()));
+ }
+ mutation.addCondition(c);
+ }
+ break;
+ case FILES: {
+ // ELASTICITY_TODO compare values?
+ Condition c = SetEqualityIterator.createCondition(tabletMetadata.getFiles(),
- stf -> TextUtil.getBytes(stf.getMetaUpdateDeleteText()), DataFileColumnFamily.NAME);
++ stf -> stf.getMetadata().getBytes(UTF_8), DataFileColumnFamily.NAME);
+ mutation.addCondition(c);
+ }
+ break;
+ case SELECTED: {
+ Condition c =
+ new Condition(SELECTED_COLUMN.getColumnFamily(), SELECTED_COLUMN.getColumnQualifier());
+ if (tabletMetadata.getSelectedFiles() != null) {
+ // ensure the SelectedFiles metadata value is re-encoded in case it was manually edited
+ c.setIterators(new IteratorSetting(INITIAL_ITERATOR_PRIO, SortedFilesIterator.class));
+ c = c.setValue(
+ Objects.requireNonNull(tabletMetadata.getSelectedFiles().getMetadataValue()));
+ }
+ mutation.addCondition(c);
+ }
+ break;
+ case ECOMP: {
+ Condition c =
+ SetEqualityIterator.createCondition(tabletMetadata.getExternalCompactions().keySet(),
+ ecid -> ecid.canonical().getBytes(UTF_8), ExternalCompactionColumnFamily.NAME);
+ mutation.addCondition(c);
+ }
+ break;
+ case LOCATION:
+ if (tabletMetadata.getLocation() == null) {
+ requireAbsentLocation();
+ } else {
+ requireLocation(tabletMetadata.getLocation());
+ }
+ break;
+ case LOADED: {
+ Condition c = SetEqualityIterator.createCondition(tabletMetadata.getLoaded().keySet(),
- stf -> TextUtil.getBytes(stf.getMetaUpdateDeleteText()), BulkFileColumnFamily.NAME);
++ stf -> stf.getMetadata().getBytes(UTF_8), BulkFileColumnFamily.NAME);
+ mutation.addCondition(c);
+ }
+ break;
+ case COMPACTED: {
+ Condition c = SetEqualityIterator.createCondition(tabletMetadata.getCompacted(),
+ ftid -> FateTxId.formatTid(ftid).getBytes(UTF_8), CompactedColumnFamily.NAME);
+ mutation.addCondition(c);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException("Column type " + type + " is not supported.");
+ }
+ }
+
+ @Override
+ public Ample.ConditionalTabletMutator requireSame(TabletMetadata tabletMetadata, ColumnType type,
+ ColumnType... otherTypes) {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ requireSameSingle(tabletMetadata, type);
+ for (var ct : otherTypes) {
+ requireSameSingle(tabletMetadata, ct);
+ }
+ return this;
+ }
+
+ @Override
+ public void submit(Ample.RejectionHandler rejectionCheck) {
+ Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate.");
+ Preconditions.checkState(sawOperationRequirement, "No operation requirements were seen");
+ getMutation();
+ mutationConsumer.accept(mutation);
+ rejectionHandlerConsumer.accept(extent, rejectionCheck);
+ }
+}
diff --cc server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java
index 50dd5edc25,de4a1f8585..6c7e5b6dac
--- 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
@@@ -22,8 -22,10 +22,9 @@@ import static java.nio.charset.Standard
import static org.apache.accumulo.core.metadata.RootTable.ZROOT_TABLET_GC_CANDIDATES;
import static org.apache.accumulo.server.util.MetadataTableUtil.EMPTY_TEXT;
+ import java.net.URI;
import java.util.Collection;
import java.util.Iterator;
-import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
@@@ -181,11 -185,42 +183,16 @@@ public class ServerAmpleImpl extends Am
}
}
- @Override
- public void removeBulkLoadEntries(TableId tableId, long tid, Text firstSplit, Text lastSplit) {
- Preconditions.checkArgument(DataLevel.of(tableId) == DataLevel.USER);
- try (
- Scanner mscanner =
- new IsolatedScanner(context.createScanner(MetadataTable.NAME, Authorizations.EMPTY));
- BatchWriter bw = context.createBatchWriter(MetadataTable.NAME)) {
- mscanner.setRange(new KeyExtent(tableId, lastSplit, firstSplit).toMetaRange());
- mscanner.fetchColumnFamily(BulkFileColumnFamily.NAME);
-
- for (Map.Entry<Key,Value> entry : mscanner) {
- log.trace("Looking at entry {} with tid {}", entry, tid);
- long entryTid = BulkFileColumnFamily.getBulkLoadTid(entry.getValue());
- if (tid == entryTid) {
- log.trace("deleting entry {}", entry);
- Key key = entry.getKey();
- Mutation m = new Mutation(key.getRow());
- m.putDelete(key.getColumnFamily(), key.getColumnQualifier());
- bw.addMutation(m);
- }
- }
- } catch (MutationsRejectedException | TableNotFoundException e) {
- throw new IllegalStateException(e);
- }
- }
-
@Override
- public void deleteGcCandidates(DataLevel level, Collection<String> paths) {
+ public void deleteGcCandidates(DataLevel level, Collection<GcCandidate> candidates,
+ GcCandidateType type) {
if (level == DataLevel.ROOT) {
- mutateRootGcCandidates(rgcc -> rgcc.remove(paths.stream()));
+ if (type == GcCandidateType.INUSE) {
+ // Deletion of INUSE candidates is not supported in 2.1.x.
+ return;
+ }
+ mutateRootGcCandidates(rgcc -> rgcc.remove(candidates.stream()));
return;
}
diff --cc server/base/src/main/java/org/apache/accumulo/server/tablets/TabletNameGenerator.java
index 4106c17408,0000000000..802b235dce
mode 100644,000000..100644
--- a/server/base/src/main/java/org/apache/accumulo/server/tablets/TabletNameGenerator.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/tablets/TabletNameGenerator.java
@@@ -1,97 -1,0 +1,89 @@@
+/*
+ * 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.tablets;
+
+import java.util.function.Consumer;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.file.FileOperations;
+import org.apache.accumulo.core.file.FilePrefix;
+import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.spi.fs.VolumeChooserEnvironment;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.fs.VolumeChooserEnvironmentImpl;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+
+/**
+ * contains code related to generating new file path for a tablet
+ */
+public class TabletNameGenerator {
+
+ public static String createTabletDirectoryName(ServerContext context, Text endRow) {
+ if (endRow == null) {
+ return MetadataSchema.TabletsSection.ServerColumnFamily.DEFAULT_TABLET_DIR_NAME;
+ } else {
+ UniqueNameAllocator namer = context.getUniqueNameAllocator();
+ return Constants.GENERATED_TABLET_DIRECTORY_PREFIX + namer.getNextName();
+ }
+ }
+
+ public static String chooseTabletDir(ServerContext context, KeyExtent extent, String dirName,
+ Consumer<String> dirCreator) {
+ VolumeChooserEnvironment chooserEnv =
+ new VolumeChooserEnvironmentImpl(extent.tableId(), extent.endRow(), context);
+ String dirUri = context.getVolumeManager().choose(chooserEnv, context.getBaseUris())
+ + Constants.HDFS_TABLES_DIR + Path.SEPARATOR + extent.tableId() + Path.SEPARATOR + dirName;
+ dirCreator.accept(dirUri);
+ return dirUri;
+ }
+
+ public static ReferencedTabletFile getNextDataFilename(FilePrefix prefix, ServerContext context,
+ KeyExtent extent, String dirName, Consumer<String> dirCreator) {
+ String extension =
+ FileOperations.getNewFileExtension(context.getTableConfiguration(extent.tableId()));
+ return new ReferencedTabletFile(
+ new Path(chooseTabletDir(context, extent, dirName, dirCreator) + "/" + prefix.toPrefix()
+ + context.getUniqueNameAllocator().getNextName() + "." + extension));
+ }
+
- public static ReferencedTabletFile getNextDataFilenameForMajc(boolean propagateDeletes,
- ServerContext context, KeyExtent extent, String dirName, Consumer<String> dirCreator) {
- String tmpFileName = getNextDataFilename(
- !propagateDeletes ? FilePrefix.MAJOR_COMPACTION_ALL_FILES : FilePrefix.MAJOR_COMPACTION,
- context, extent, dirName, dirCreator).getMetaInsert() + "_tmp";
- return new ReferencedTabletFile(new Path(tmpFileName));
- }
-
+ public static ReferencedTabletFile getNextDataFilenameForMajc(boolean propagateDeletes,
+ ServerContext context, TabletMetadata tabletMetadata, Consumer<String> dirCreator) {
+ String tmpFileName = getNextDataFilename(
+ !propagateDeletes ? FilePrefix.MAJOR_COMPACTION_ALL_FILES : FilePrefix.MAJOR_COMPACTION,
- context, tabletMetadata.getExtent(), tabletMetadata.getDirName(), dirName -> {})
- .getMetaInsert() + "_tmp";
++ context, tabletMetadata.getExtent(), tabletMetadata.getDirName(), dirName -> {}).insert()
++ .getMetadataPath() + "_tmp";
+ return new ReferencedTabletFile(new Path(tmpFileName));
+ }
+
+ public static ReferencedTabletFile computeCompactionFileDest(ReferencedTabletFile tmpFile) {
- String newFilePath = tmpFile.getMetaInsert();
++ String newFilePath = tmpFile.getNormalizedPathStr();
+ int idx = newFilePath.indexOf("_tmp");
+ if (idx > 0) {
+ newFilePath = newFilePath.substring(0, idx);
+ } else {
- throw new IllegalArgumentException(
- "Expected compaction tmp file " + tmpFile.getMetaInsert() + " to have suffix '_tmp'");
++ throw new IllegalArgumentException("Expected compaction tmp file "
++ + tmpFile.getNormalizedPathStr() + " to have suffix '_tmp'");
+ }
+ return new ReferencedTabletFile(new Path(newFilePath));
+ }
+}
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
index 852bdea000,1ac0012a53..cad4381c53
--- a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
@@@ -50,6 -49,6 +50,7 @@@ import org.apache.accumulo.core.client.
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.client.admin.TabletHostingGoal;
import org.apache.accumulo.core.client.admin.TimeType;
import org.apache.accumulo.core.clientImpl.BatchWriterImpl;
import org.apache.accumulo.core.clientImpl.ScannerImpl;
@@@ -166,12 -172,12 +166,13 @@@ public class MetadataTableUtil
}
public static void addTablet(KeyExtent extent, String path, ServerContext context,
-- TimeType timeType, ServiceLock zooLock) {
++ TimeType timeType, ServiceLock zooLock, TabletHostingGoal goal) {
TabletMutator tablet = context.getAmple().mutateTablet(extent);
tablet.putPrevEndRow(extent.prevEndRow());
tablet.putDirName(path);
tablet.putTime(new MetadataTime(0, timeType));
- tablet.putZooLock(zooLock);
+ tablet.putZooLock(context.getZooKeeperRoot(), zooLock);
++ tablet.putHostingGoal(goal);
tablet.mutate();
}
diff --cc server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java
index a206bbb4a2,1aeadb6736..f25361d8a9
--- a/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java
@@@ -22,13 -22,13 +22,16 @@@ import static org.junit.jupiter.api.Ass
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+ import java.net.URI;
import java.util.List;
+import java.util.Set;
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.fate.FateTxId;
import org.apache.accumulo.core.metadata.MetadataTable;
++import org.apache.accumulo.core.metadata.ReferencedTabletFile;
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.BulkFileColumnFamily;
@@@ -38,8 -37,8 +41,9 @@@ import org.apache.accumulo.core.metadat
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
+import org.apache.accumulo.core.metadata.schema.SelectedFiles;
import org.apache.accumulo.server.ServerContext;
-import org.apache.accumulo.server.zookeeper.TransactionWatcher.Arbitrator;
++import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.easymock.EasyMock;
import org.junit.jupiter.api.Test;
@@@ -135,11 -155,34 +139,17 @@@ public class MetadataConstraintsTest
Mutation m;
List<Short> violations;
- // inactive txid
- m = new Mutation(new Text("0;foo"));
- m.put(BulkFileColumnFamily.NAME,
- StoredTabletFile
- .of(URI.create("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile"), new Range())
- .getMetadataText(),
- new Value("12345"));
- m.put(DataFileColumnFamily.NAME,
- StoredTabletFile
- .of(URI.create("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile"), new Range())
- .getMetadataText(),
- new DataFileValue(1, 1).encodeAsValue());
- violations = mc.check(createEnv(), m);
- assertNotNull(violations);
- assertEquals(1, violations.size());
- assertEquals(Short.valueOf((short) 8), violations.get(0));
-
// txid that throws exception
m = new Mutation(new Text("0;foo"));
- m.put(BulkFileColumnFamily.NAME, new Text("hdfs://nn1/a/accumulo/tables/2b/t-001/someFile"),
+ m.put(BulkFileColumnFamily.NAME,
+ StoredTabletFile
+ .of(URI.create("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile"), new Range())
+ .getMetadataText(),
- new Value("9"));
+ new Value("bad value"));
- m.put(DataFileColumnFamily.NAME, new Text("hdfs://nn1/a/accumulo/tables/2b/t-001/someFile"),
+ m.put(DataFileColumnFamily.NAME,
+ StoredTabletFile
+ .of(URI.create("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile"), new Range())
+ .getMetadataText(),
new DataFileValue(1, 1).encodeAsValue());
violations = mc.check(createEnv(), m);
assertNotNull(violations);
@@@ -246,14 -345,15 +312,15 @@@
// Missing beginning of path
m = new Mutation(new Text("0;foo"));
- m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new Value("5"));
+ m.put(BulkFileColumnFamily.NAME, new Text(StoredTabletFile
+ .of(URI.create("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile"), new Range())
+ .getMetadata().replace("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile", "/someFile")),
+ new Value("5"));
violations = mc.check(createEnv(), m);
assertNotNull(violations);
- assertEquals(2, violations.size());
+ assertEquals(1, violations.size());
- assertEquals(Short.valueOf((short) 9), violations.get(0));
+ assertEquals(Short.valueOf((short) 12), violations.get(0));
assertNotNull(mc.getViolationDescription(violations.get(0)));
- // No DataFileColumnFamily included
- assertEquals(Short.valueOf((short) 8), violations.get(1));
// Missing tables directory in path
m = new Mutation(new Text("0;foo"));
@@@ -261,10 -365,21 +332,20 @@@
new Value("5"));
violations = mc.check(createEnv(), m);
assertNotNull(violations);
- assertEquals(2, violations.size());
+ assertEquals(1, violations.size());
- assertEquals(Short.valueOf((short) 9), violations.get(0));
+ assertEquals(Short.valueOf((short) 12), violations.get(0));
- // No DataFileColumnFamily included
- assertEquals(Short.valueOf((short) 8), violations.get(1));
+
- // No DataFileColumnFamily included
+ m = new Mutation(new Text("0;foo"));
+ m.put(BulkFileColumnFamily.NAME,
+ StoredTabletFile
+ .of(URI.create("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile"), new Range())
+ .getMetadataText(),
+ new Value("5"));
+ violations = mc.check(createEnv(), m);
+ assertNotNull(violations);
+ assertEquals(1, violations.size());
+ assertEquals(Short.valueOf((short) 8), violations.get(0));
+ assertNotNull(mc.getViolationDescription(violations.get(0)));
}
@@@ -285,7 -400,19 +366,19 @@@
// Missing beginning of path
m = new Mutation(new Text("0;foo"));
- m.put(columnFamily, new Text("/someFile"), value);
+ m.put(columnFamily, new Text(StoredTabletFile
+ .of(URI.create("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile"), new Range())
+ .getMetadata().replace("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile", "/someFile")),
+ value);
+ violations = mc.check(createEnv(), m);
+ assertNotNull(violations);
+ assertEquals(1, violations.size());
- assertEquals(Short.valueOf((short) 9), violations.get(0));
++ assertEquals(Short.valueOf((short) 12), violations.get(0));
+ assertNotNull(mc.getViolationDescription(violations.get(0)));
+
+ // Bad Json - only path so should fail parsing
+ m = new Mutation(new Text("0;foo"));
+ m.put(columnFamily, new Text("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile"), value);
violations = mc.check(createEnv(), m);
assertNotNull(violations);
assertEquals(1, violations.size());
@@@ -308,67 -442,7 +408,66 @@@
violations = mc.check(createEnv(), m);
assertNull(violations);
- assertNotNull(mc.getViolationDescription((short) 9));
+ assertNotNull(mc.getViolationDescription((short) 12));
+ }
+
+ @Test
+ public void testOperationId() {
+ MetadataConstraints mc = new MetadataConstraints();
+ Mutation m;
+ List<Short> violations;
+
+ m = new Mutation(new Text("0;foo"));
+ ServerColumnFamily.OPID_COLUMN.put(m, new Value("bad id"));
+ violations = mc.check(createEnv(), m);
+ assertNotNull(violations);
+ assertEquals(1, violations.size());
+ assertEquals(Short.valueOf((short) 9), violations.get(0));
+
+ m = new Mutation(new Text("0;foo"));
+ ServerColumnFamily.OPID_COLUMN.put(m, new Value("MERGING:FATE[123abc]"));
+ violations = mc.check(createEnv(), m);
+ assertNull(violations);
+ }
+
+ @Test
+ public void testSelectedFiles() {
+ MetadataConstraints mc = new MetadataConstraints();
+ Mutation m;
+ List<Short> violations;
+
+ m = new Mutation(new Text("0;foo"));
+ ServerColumnFamily.SELECTED_COLUMN.put(m, new Value("bad id"));
+ violations = mc.check(createEnv(), m);
+ assertNotNull(violations);
+ assertEquals(1, violations.size());
+ assertEquals(Short.valueOf((short) 11), violations.get(0));
+
+ m = new Mutation(new Text("0;foo"));
+ ServerColumnFamily.SELECTED_COLUMN.put(m,
- new Value(new SelectedFiles(
- Set.of(StoredTabletFile
- .of("hdfs://nn.somewhere.com:86753/accumulo/tables/42/t-0000/F00001.rf")),
- true, 42L).getMetadataValue()));
++ new Value(new SelectedFiles(Set.of(new ReferencedTabletFile(
++ new Path("hdfs://nn.somewhere.com:86753/accumulo/tables/42/t-0000/F00001.rf"))
++ .insert()), true, 42L).getMetadataValue()));
+ violations = mc.check(createEnv(), m);
+ assertNull(violations);
}
+ @Test
+ public void testCompacted() {
+ MetadataConstraints mc = new MetadataConstraints();
+ Mutation m;
+ List<Short> violations;
+
+ m = new Mutation(new Text("0;foo"));
+ m.put(CompactedColumnFamily.STR_NAME, FateTxId.formatTid(45), "");
+ violations = mc.check(createEnv(), m);
+ assertNull(violations);
+
+ m = new Mutation(new Text("0;foo"));
+ m.put(CompactedColumnFamily.STR_NAME, "incorrect data", "");
+ violations = mc.check(createEnv(), m);
+ assertNotNull(violations);
+ assertEquals(1, violations.size());
+ assertEquals(Short.valueOf((short) 13), violations.get(0));
+ }
}
diff --cc server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletManagementTest.java
index e09cca2b34,0000000000..b9fd8837d7
mode 100644,000000..100644
--- a/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletManagementTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletManagementTest.java
@@@ -1,163 -1,0 +1,171 @@@
+/*
+ * 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.manager.state;
+
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.COMPACT_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.FLUSH_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.TIME_COLUMN;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+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.fate.FateTxId;
+import org.apache.accumulo.core.iterators.user.WholeRowIterator;
+import org.apache.accumulo.core.manager.state.TabletManagement;
+import org.apache.accumulo.core.manager.state.TabletManagement.ManagementAction;
++import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.StoredTabletFile;
+import org.apache.accumulo.core.metadata.schema.DataFileValue;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.BulkFileColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ClonedColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CurrentLocationColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LastLocationColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
++import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.Test;
+
+public class TabletManagementTest {
+
+ private SortedMap<Key,Value> toRowMap(Mutation mutation) {
+ SortedMap<Key,Value> rowMap = new TreeMap<>();
+ mutation.getUpdates().forEach(cu -> {
+ Key k = new Key(mutation.getRow(), cu.getColumnFamily(), cu.getColumnQualifier(),
+ cu.getTimestamp());
+ Value v = new Value(cu.getValue());
+ rowMap.put(k, v);
+ });
+ return rowMap;
+ }
+
+ private SortedMap<Key,Value> createMetadataEntryKV(KeyExtent extent) {
+
+ Mutation mutation = TabletColumnFamily.createPrevRowMutation(extent);
+
+ COMPACT_COLUMN.put(mutation, new Value("5"));
+ DIRECTORY_COLUMN.put(mutation, new Value("t-0001757"));
+ FLUSH_COLUMN.put(mutation, new Value("6"));
+ TIME_COLUMN.put(mutation, new Value("M123456789"));
+
- String bf1 = "hdfs://nn1/acc/tables/1/t-0001/bf1";
- String bf2 = "hdfs://nn1/acc/tables/1/t-0001/bf2";
- mutation.at().family(BulkFileColumnFamily.NAME).qualifier(bf1).put(FateTxId.formatTid(56));
- mutation.at().family(BulkFileColumnFamily.NAME).qualifier(bf2).put(FateTxId.formatTid(59));
++ StoredTabletFile bf1 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/bf1")).insert();
++ StoredTabletFile bf2 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/bf2")).insert();
++ mutation.at().family(BulkFileColumnFamily.NAME).qualifier(bf1.getMetadata())
++ .put(FateTxId.formatTid(56));
++ mutation.at().family(BulkFileColumnFamily.NAME).qualifier(bf2.getMetadata())
++ .put(FateTxId.formatTid(59));
+
+ mutation.at().family(ClonedColumnFamily.NAME).qualifier("").put("OK");
+
+ DataFileValue dfv1 = new DataFileValue(555, 23);
- StoredTabletFile tf1 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-0001/df1.rf");
- StoredTabletFile tf2 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-0001/df2.rf");
- mutation.at().family(DataFileColumnFamily.NAME).qualifier(tf1.getMetaUpdateDelete())
- .put(dfv1.encode());
++ StoredTabletFile tf1 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/df1.rf")).insert();
++ StoredTabletFile tf2 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/df2.rf")).insert();
++ mutation.at().family(DataFileColumnFamily.NAME).qualifier(tf1.getMetadata()).put(dfv1.encode());
+ DataFileValue dfv2 = new DataFileValue(234, 13);
- mutation.at().family(DataFileColumnFamily.NAME).qualifier(tf2.getMetaUpdateDelete())
- .put(dfv2.encode());
++ mutation.at().family(DataFileColumnFamily.NAME).qualifier(tf2.getMetadata()).put(dfv2.encode());
+
+ mutation.at().family(CurrentLocationColumnFamily.NAME).qualifier("s001").put("server1:8555");
+
+ mutation.at().family(LastLocationColumnFamily.NAME).qualifier("s000").put("server2:8555");
+
+ LogEntry le1 = new LogEntry(extent, 55, "lf1");
+ mutation.at().family(le1.getColumnFamily()).qualifier(le1.getColumnQualifier())
+ .timestamp(le1.timestamp).put(le1.getValue());
+ LogEntry le2 = new LogEntry(extent, 57, "lf2");
+ mutation.at().family(le2.getColumnFamily()).qualifier(le2.getColumnQualifier())
+ .timestamp(le2.timestamp).put(le2.getValue());
+
- StoredTabletFile sf1 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-0001/sf1.rf");
- StoredTabletFile sf2 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-0001/sf2.rf");
- mutation.at().family(ScanFileColumnFamily.NAME).qualifier(sf1.getMetaUpdateDelete()).put("");
- mutation.at().family(ScanFileColumnFamily.NAME).qualifier(sf2.getMetaUpdateDelete()).put("");
++ StoredTabletFile sf1 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/sf1.rf")).insert();
++ StoredTabletFile sf2 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-0001/sf2.rf")).insert();
++ mutation.at().family(ScanFileColumnFamily.NAME).qualifier(sf1.getMetadata()).put("");
++ mutation.at().family(ScanFileColumnFamily.NAME).qualifier(sf2.getMetadata()).put("");
+
+ return toRowMap(mutation);
+
+ }
+
+ @Test
+ public void testEncodeDecodeWithReasons() throws Exception {
+ KeyExtent extent = new KeyExtent(TableId.of("5"), new Text("df"), new Text("da"));
+
+ final Set<ManagementAction> actions =
+ Set.of(ManagementAction.NEEDS_LOCATION_UPDATE, ManagementAction.NEEDS_SPLITTING);
+
+ final SortedMap<Key,Value> entries = createMetadataEntryKV(extent);
+
+ TabletManagement.addActions(entries, actions);
+ Key key = entries.firstKey();
+ Value val = WholeRowIterator.encodeRow(new ArrayList<>(entries.keySet()),
+ new ArrayList<>(entries.values()));
+
+ // Remove the REASONS column from the entries map for the comparison check
+ // below
+ entries.remove(new Key(key.getRow().toString(), "REASONS", ""));
+
+ TabletManagement tmi = new TabletManagement(key, val, true);
+ assertEquals(entries, tmi.getTabletMetadata().getKeyValues());
+ assertEquals(actions, tmi.getActions());
+ }
+
+ @Test
+ public void testBinary() throws Exception {
+ // test end row with non ascii data
+ Text endRow = new Text(new byte[] {'m', (byte) 0xff});
+ KeyExtent extent = new KeyExtent(TableId.of("5"), endRow, new Text("da"));
+
+ final Set<ManagementAction> actions =
+ Set.of(ManagementAction.NEEDS_LOCATION_UPDATE, ManagementAction.NEEDS_SPLITTING);
+
+ final SortedMap<Key,Value> entries = createMetadataEntryKV(extent);
+
+ TabletManagement.addActions(entries, actions);
+ Key key = entries.firstKey();
+ Value val = WholeRowIterator.encodeRow(new ArrayList<>(entries.keySet()),
+ new ArrayList<>(entries.values()));
+
+ assertTrue(entries.keySet().stream().allMatch(k -> k.getRow().equals(extent.toMetaRow())));
+
+ // Remove the REASONS column from the entries map for the comparison check
+ // below
+ entries.remove(new Key(key.getRow(), new Text("REASONS"), new Text("")));
+
+ TabletManagement tmi = new TabletManagement(key, val, true);
+ assertEquals(entries, tmi.getTabletMetadata().getKeyValues());
+ assertEquals(actions, tmi.getActions());
+
+ }
+}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
index 78a769b5f9,d59ca8d552..0f539e199f
--- a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
@@@ -98,10 -94,9 +98,9 @@@ import org.apache.accumulo.core.manager
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.TServerInstance;
- import org.apache.accumulo.core.metadata.TabletState;
-import org.apache.accumulo.core.metadata.TabletLocationState;
import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metrics.MetricsUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.spi.balancer.BalancerEnvironment;
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/TabletGroupWatcher.java
index 0bdc4b0dc0,21ae3ae796..df7fee1e21
--- a/server/manager/src/main/java/org/apache/accumulo/manager/TabletGroupWatcher.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/TabletGroupWatcher.java
@@@ -81,13 -80,13 +85,16 @@@ import org.apache.accumulo.core.metadat
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataTime;
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.metadata.schema.TabletsMetadata;
import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.spi.balancer.data.TabletServerId;
+ import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.core.util.TextUtil;
import org.apache.accumulo.core.util.threads.Threads.AccumuloDaemonThread;
import org.apache.accumulo.manager.Manager.TabletGoalState;
+import org.apache.accumulo.manager.split.SplitTask;
import org.apache.accumulo.manager.state.MergeStats;
import org.apache.accumulo.manager.state.TableCounts;
import org.apache.accumulo.manager.state.TableStats;
@@@ -112,31 -108,16 +119,34 @@@ import org.apache.accumulo.server.util.
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+ import com.google.common.annotations.VisibleForTesting;
+ import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSortedSet;
import com.google.common.collect.Iterators;
+ import com.google.common.collect.Sets;
abstract class TabletGroupWatcher extends AccumuloDaemonThread {
- // Constants used to make sure assignment logging isn't excessive in quantity or size
+ public static class BadLocationStateException extends Exception {
+ private static final long serialVersionUID = 2L;
+
+ // store as byte array because Text isn't Serializable
+ private final byte[] metadataTableEntry;
+
+ public BadLocationStateException(String msg, Text row) {
+ super(msg);
+ this.metadataTableEntry = TextUtil.getBytes(requireNonNull(row));
+ }
+
+ public Text getEncodedEndRow() {
+ return new Text(metadataTableEntry);
+ }
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(TabletGroupWatcher.class);
private final Manager manager;
private final TabletStateStore store;
private final TabletGroupWatcher dependentWatcher;
@@@ -301,25 -252,15 +311,24 @@@
var mStats = currentMerges.get(k);
return mStats != null ? mStats : new MergeStats(new MergeInfo());
});
- TabletGoalState goal = manager.getGoalState(tls, mergeStats.getMergeInfo());
- Location location = tls.getLocation();
- TabletState state = tls.getState(currentTServers.keySet());
-
- TabletLogger.missassigned(tls.extent, goal.toString(), state.toString(),
- tls.getFutureServer(), tls.getCurrentServer(), tls.walogs.size());
+ TabletGoalState goal = manager.getGoalState(tm, mergeStats.getMergeInfo());
+ TabletState state = TabletState.compute(tm, currentTServers.keySet(),
+ manager.tabletBalancer, resourceGroups);
+
+ final Location location = tm.getLocation();
+ Location current = null;
+ Location future = null;
+ if (tm.hasCurrent()) {
+ current = tm.getLocation();
+ } else {
+ future = tm.getLocation();
+ }
+ TabletLogger.missassigned(tm.getExtent(), goal.toString(), state.toString(),
+ future != null ? future.getServerInstance() : null,
+ current != null ? current.getServerInstance() : null, tm.getLogs().size());
stats.update(tableId, state);
- mergeStats.update(tm.getExtent(), state, tm.hasChopped(), !tm.getLogs().isEmpty());
- sendChopRequest(mergeStats.getMergeInfo(), state, tm);
- mergeStats.update(tls.extent, state);
++ mergeStats.update(tm.getExtent(), state);
// Always follow through with assignments
if (state == TabletState.ASSIGNED) {
@@@ -329,10 -268,10 +338,10 @@@
}
if (Manager.log.isTraceEnabled()) {
Manager.log.trace(
-- "[{}] Shutting down all Tservers: {}, dependentCount: {} Extent: {}, state: {}, goal: {}",
++ "[{}] Shutting down all Tservers: {}, dependentCount: {} Extent: {}, state: {}, goal: {} actions:{}",
store.name(), manager.serversToShutdown.equals(currentTServers.keySet()),
- dependentWatcher == null ? "null" : dependentWatcher.assignedOrHosted(), tls.extent,
- state, goal);
+ dependentWatcher == null ? "null" : dependentWatcher.assignedOrHosted(),
- tm.getExtent(), state, goal);
++ tm.getExtent(), state, goal, actions);
}
// if we are shutting down all the tabletservers, we have to do it in order
@@@ -371,101 -310,65 +380,102 @@@
}
}
- if (goal == TabletGoalState.HOSTED) {
- if ((state != TabletState.HOSTED && !tls.walogs.isEmpty())
- && manager.recoveryManager.recoverLogs(tls.extent, tls.walogs)) {
- continue;
- }
- switch (state) {
- case HOSTED:
- if (location.getServerInstance().equals(manager.migrations.get(tls.extent))) {
- manager.migrations.remove(tls.extent);
- }
- break;
- case ASSIGNED_TO_DEAD_SERVER:
- hostDeadTablet(tLists, tls, location, wals);
- break;
- case SUSPENDED:
- hostSuspendedTablet(tLists, tls, location, tableConf);
- break;
- case UNASSIGNED:
- hostUnassignedTablet(tLists, tls.extent, new UnassignedTablet(location, tls.last));
- break;
- case ASSIGNED:
- // Send another reminder
- tLists.assigned.add(new Assignment(tls.extent, tls.getFutureServer(), tls.last));
- break;
+ if (actions.contains(ManagementAction.NEEDS_SPLITTING)) {
+ LOG.debug("{} may need splitting.", tm.getExtent());
+ if (manager.getSplitter().isSplittable(tm)) {
+ if (manager.getSplitter().addSplitStarting(tm.getExtent())) {
+ LOG.debug("submitting tablet {} for split", tm.getExtent());
+ manager.getSplitter()
+ .executeSplit(new SplitTask(manager.getContext(), tm, manager));
+ }
+ } else {
+ LOG.debug("{} is not splittable.", tm.getExtent());
}
- } else {
- switch (state) {
- case SUSPENDED:
- // Request a move to UNASSIGNED, so as to allow balancing to continue.
- tLists.suspendedToGoneServers.add(tls);
- cancelOfflineTableMigrations(tls.extent);
- break;
- case UNASSIGNED:
- cancelOfflineTableMigrations(tls.extent);
- break;
- case ASSIGNED_TO_DEAD_SERVER:
- unassignDeadTablet(tLists, tls, wals);
- break;
- case HOSTED:
- TServerConnection client =
- manager.tserverSet.getConnection(location.getServerInstance());
- if (client != null) {
- Manager.log.trace("[{}] Requesting TabletServer {} unload {} {}", store.name(),
- location.getServerInstance(), tls.extent, goal.howUnload());
- client.unloadTablet(manager.managerLock, tls.extent, goal.howUnload(),
- manager.getSteadyTime());
- unloaded++;
- totalUnloaded++;
- } else {
- Manager.log.warn("Could not connect to server {}", location);
- }
- break;
- case ASSIGNED:
- break;
+ // ELASITICITY_TODO: See #3605. Merge is non-functional. Left this commented out code to
+ // show where merge used to make a call to split a tablet.
+ // sendSplitRequest(mergeStats.getMergeInfo(), state, tm);
+ }
+
+ if (actions.contains(ManagementAction.NEEDS_COMPACTING)) {
+ var jobs = compactionGenerator.generateJobs(tm,
+ TabletManagementIterator.determineCompactionKinds(actions));
+ LOG.debug("{} may need compacting adding {} jobs", tm.getExtent(), jobs.size());
+ manager.getCompactionQueues().add(tm, jobs);
+ }
+
+ // ELASITICITY_TODO the case where a planner generates compactions at time T1 for tablet
+ // and later at time T2 generates nothing for the same tablet is not being handled. At
+ // time T1 something could have been queued. However at time T2 we will not clear those
+ // entries from the queue because we see nothing here for that case. After a full
+ // metadata scan could remove any tablets that were not updated during the scan.
+
- if (actions.contains(ManagementAction.NEEDS_LOCATION_UPDATE)) {
++ if (actions.contains(ManagementAction.NEEDS_LOCATION_UPDATE)
++ || actions.contains(ManagementAction.IS_MERGING)) {
+ if (goal == TabletGoalState.HOSTED) {
+ if ((state != TabletState.HOSTED && !tm.getLogs().isEmpty())
+ && manager.recoveryManager.recoverLogs(tm.getExtent(), tm.getLogs())) {
+ continue;
+ }
+ switch (state) {
+ case HOSTED:
+ if (location.getServerInstance().equals(manager.migrations.get(tm.getExtent()))) {
+ manager.migrations.remove(tm.getExtent());
+ }
+ break;
+ case ASSIGNED_TO_DEAD_SERVER:
+ hostDeadTablet(tLists, tm, location, wals);
+ break;
+ case SUSPENDED:
+ hostSuspendedTablet(tLists, tm, location, tableConf);
+ break;
+ case UNASSIGNED:
+ hostUnassignedTablet(tLists, tm.getExtent(),
+ new UnassignedTablet(location, tm.getLast()));
+ break;
+ case ASSIGNED:
+ // Send another reminder
+ tLists.assigned.add(new Assignment(tm.getExtent(),
+ future != null ? future.getServerInstance() : null, tm.getLast()));
+ break;
+ default:
+ break;
+ }
+ } else {
+ switch (state) {
+ case SUSPENDED:
+ // Request a move to UNASSIGNED, so as to allow balancing to continue.
+ tLists.suspendedToGoneServers.add(tm);
+ cancelOfflineTableMigrations(tm.getExtent());
+ break;
+ case UNASSIGNED:
+ cancelOfflineTableMigrations(tm.getExtent());
+ break;
+ case ASSIGNED_TO_DEAD_SERVER:
+ unassignDeadTablet(tLists, tm, wals);
+ break;
+ case NEEDS_REASSIGNMENT:
+ case HOSTED:
+ TServerConnection client =
+ manager.tserverSet.getConnection(location.getServerInstance());
+ if (client != null) {
+ client.unloadTablet(manager.managerLock, tm.getExtent(), goal.howUnload(),
+ manager.getSteadyTime());
+ unloaded++;
+ totalUnloaded++;
+ } else {
+ Manager.log.warn("Could not connect to server {}", location);
+ }
+ break;
+ case ASSIGNED:
+ break;
+ }
}
+ counts[state.ordinal()]++;
}
- counts[state.ordinal()]++;
}
+ // ELASTICITY_TODO: Add handling for other actions
+
flushChanges(tLists, wals);
// provide stats after flushing changes to avoid race conditions w/ delete table
@@@ -770,6 -716,6 +823,7 @@@
Manager.log.debug("Deleting tablets for {}", extent);
MetadataTime metadataTime = null;
KeyExtent followingTablet = null;
++ Set<TabletHostingGoal> goals = new HashSet<>();
if (extent.endRow() != null) {
Key nextExtent = new Key(extent.endRow()).followingKey(PartialKey.ROW);
followingTablet =
@@@ -791,6 -737,6 +845,7 @@@
scanner.setRange(deleteRange);
ServerColumnFamily.DIRECTORY_COLUMN.fetch(scanner);
ServerColumnFamily.TIME_COLUMN.fetch(scanner);
++ HostingColumnFamily.GOAL_COLUMN.fetch(scanner);
scanner.fetchColumnFamily(DataFileColumnFamily.NAME);
scanner.fetchColumnFamily(CurrentLocationColumnFamily.NAME);
Set<ReferenceFile> datafilesAndDirs = new TreeSet<>();
@@@ -816,6 -762,6 +871,9 @@@
ample.putGcFileAndDirCandidates(extent.tableId(), datafilesAndDirs);
datafilesAndDirs.clear();
}
++ } else if (HostingColumnFamily.GOAL_COLUMN.hasColumns(key)) {
++ TabletHostingGoal thisGoal = TabletHostingGoalUtil.fromValue(entry.getValue());
++ goals.add(thisGoal);
}
}
ample.putGcFileAndDirCandidates(extent.tableId(), datafilesAndDirs);
@@@ -843,7 -788,7 +900,7 @@@
// Recreate the default tablet to hold the end of the table
MetadataTableUtil.addTablet(new KeyExtent(extent.tableId(), null, extent.prevEndRow()),
ServerColumnFamily.DEFAULT_TABLET_DIR_NAME, manager.getContext(),
-- metadataTime.getType(), manager.managerLock);
++ metadataTime.getType(), manager.managerLock, getMergeHostingGoal(extent, goals));
}
} catch (RuntimeException | TableNotFoundException ex) {
throw new AccumuloException(ex);
@@@ -899,10 -906,9 +1020,12 @@@
} else if (ServerColumnFamily.DIRECTORY_COLUMN.hasColumns(key)) {
var allVolumesDir = new AllVolumesDirectory(range.tableId(), value.toString());
bw.addMutation(manager.getContext().getAmple().createDeleteMutation(allVolumesDir));
+ } else if (HostingColumnFamily.GOAL_COLUMN.hasColumns(key)) {
+ TabletHostingGoal thisGoal = TabletHostingGoalUtil.fromValue(value);
+ goals.add(thisGoal);
}
+
+ lastExtent = keyExtent;
}
// read the logical time from the last tablet in the merge range, it is not included in
@@@ -932,16 -934,6 +1055,11 @@@
// delete any entries for external compactions
extCompIds.forEach(ecid -> m.putDelete(ExternalCompactionColumnFamily.STR_NAME, ecid));
+ // Set the TabletHostingGoal for this tablet based on the goals of the other tablets in
+ // the merge range. Always takes priority over never.
- TabletHostingGoal mergeHostingGoal = TabletHostingGoal.ONDEMAND;
- if (range.isMeta() || goals.contains(TabletHostingGoal.ALWAYS)) {
- mergeHostingGoal = TabletHostingGoal.ALWAYS;
- } else if (goals.contains(TabletHostingGoal.NEVER)) {
- mergeHostingGoal = TabletHostingGoal.NEVER;
- }
++ TabletHostingGoal mergeHostingGoal = getMergeHostingGoal(range, goals);
+ HostingColumnFamily.GOAL_COLUMN.put(m, TabletHostingGoalUtil.toValue(mergeHostingGoal));
+
if (!m.getUpdates().isEmpty()) {
bw.addMutation(m);
}
@@@ -964,12 -956,202 +1082,213 @@@
deleteTablets(info, scanRange, bw, client);
- // Clean-up the last chopped marker
- var m2 = new Mutation(stopRow);
- ChoppedColumnFamily.CHOPPED_COLUMN.putDelete(m2);
- bw.addMutation(m2);
+ } catch (Exception ex) {
+ throw new AccumuloException(ex);
+ }
+ }
+
++ private static TabletHostingGoal getMergeHostingGoal(KeyExtent range,
++ Set<TabletHostingGoal> goals) {
++ TabletHostingGoal mergeHostingGoal = TabletHostingGoal.ONDEMAND;
++ if (range.isMeta() || goals.contains(TabletHostingGoal.ALWAYS)) {
++ mergeHostingGoal = TabletHostingGoal.ALWAYS;
++ } else if (goals.contains(TabletHostingGoal.NEVER)) {
++ mergeHostingGoal = TabletHostingGoal.NEVER;
++ }
++ return mergeHostingGoal;
++ }
++
+ // This method is used to detect if a tablet needs to be split/chopped for a delete
+ // Instead of performing a split or chop compaction, the tablet will have its files fenced.
+ private boolean needsFencingForDeletion(MergeInfo info, KeyExtent keyExtent) {
+ // Does this extent cover the end points of the delete?
+ final Predicate<Text> isWithin = r -> r != null && keyExtent.contains(r);
+ final Predicate<Text> isNotBoundary =
+ r -> !r.equals(keyExtent.endRow()) && !r.equals(keyExtent.prevEndRow());
+ final KeyExtent deleteRange = info.getExtent();
+
+ return (keyExtent.overlaps(deleteRange) && Stream
+ .of(deleteRange.prevEndRow(), deleteRange.endRow()).anyMatch(isWithin.and(isNotBoundary)))
+ || info.needsToBeChopped(keyExtent);
+ }
+
+ // Instead of splitting or chopping tablets for a delete we instead create ranges
+ // to exclude the portion of the tablet that should be deleted
+ private Text followingRow(Text row) {
+ if (row == null) {
+ return null;
+ }
+ return new Key(row).followingKey(PartialKey.ROW).getRow();
+ }
+
+ // Instead of splitting or chopping tablets for a delete we instead create ranges
+ // to exclude the portion of the tablet that should be deleted
+ private List<Range> createRangesForDeletion(TabletMetadata tabletMetadata,
+ final KeyExtent deleteRange) {
+ final KeyExtent tabletExtent = tabletMetadata.getExtent();
+
+ // If the delete range wholly contains the tablet being deleted then there is no range to clip
+ // files to because the files should be completely dropped.
+ Preconditions.checkArgument(!deleteRange.contains(tabletExtent), "delete range:%s tablet:%s",
+ deleteRange, tabletExtent);
+
+ final List<Range> ranges = new ArrayList<>();
+
+ if (deleteRange.overlaps(tabletExtent)) {
+ if (deleteRange.prevEndRow() != null
+ && tabletExtent.contains(followingRow(deleteRange.prevEndRow()))) {
+ Manager.log.trace("Fencing tablet {} files to ({},{}]", tabletExtent,
+ tabletExtent.prevEndRow(), deleteRange.prevEndRow());
+ ranges.add(new Range(tabletExtent.prevEndRow(), false, deleteRange.prevEndRow(), true));
+ }
+
+ // This covers the case of when a deletion range overlaps the last tablet. We need to create a
+ // range that excludes the deletion.
+ if (deleteRange.endRow() != null
+ && tabletMetadata.getExtent().contains(deleteRange.endRow())) {
+ Manager.log.trace("Fencing tablet {} files to ({},{}]", tabletExtent, deleteRange.endRow(),
+ tabletExtent.endRow());
+ ranges.add(new Range(deleteRange.endRow(), false, tabletExtent.endRow(), true));
+ }
+ } else {
+ Manager.log.trace(
+ "Fencing tablet {} files to itself because it does not overlap delete range",
+ tabletExtent);
+ ranges.add(tabletExtent.toDataRange());
+ }
+
+ return ranges;
+ }
+
+ private Pair<KeyExtent,KeyExtent> updateMetadataRecordsForDelete(MergeInfo info)
+ throws AccumuloException {
+ final KeyExtent range = info.getExtent();
+
+ String targetSystemTable = MetadataTable.NAME;
+ if (range.isMeta()) {
+ targetSystemTable = RootTable.NAME;
+ }
+ final Pair<KeyExtent,KeyExtent> startAndEndTablets;
+
+ final AccumuloClient client = manager.getContext();
+
+ try (BatchWriter bw = client.createBatchWriter(targetSystemTable)) {
+ final Text startRow = range.prevEndRow();
+ final Text endRow = range.endRow() != null
+ ? new Key(range.endRow()).followingKey(PartialKey.ROW).getRow() : null;
+
+ // Find the tablets that overlap the start and end row of the deletion range
+ // If the startRow is null then there will be an empty startTablet we don't need
+ // to fence a starting tablet as we are deleting everything up to the end tablet
+ // Likewise, if the endRow is null there will be an empty endTablet as we are deleting
+ // all tablets after the starting tablet
+ final Optional<TabletMetadata> startTablet = Optional.ofNullable(startRow).flatMap(
+ row -> loadTabletMetadata(range.tableId(), row, ColumnType.PREV_ROW, ColumnType.FILES));
+ final Optional<TabletMetadata> endTablet = Optional.ofNullable(endRow).flatMap(
+ row -> loadTabletMetadata(range.tableId(), row, ColumnType.PREV_ROW, ColumnType.FILES));
+
+ // Store the tablets in a Map if present so that if we have the same Tablet we
+ // only need to process the same tablet once when fencing
+ final SortedMap<KeyExtent,TabletMetadata> tabletMetadatas = new TreeMap<>();
+ startTablet.ifPresent(ft -> tabletMetadatas.put(ft.getExtent(), ft));
+ endTablet.ifPresent(lt -> tabletMetadatas.putIfAbsent(lt.getExtent(), lt));
+
+ // Capture the tablets to return them or null if not loaded
+ startAndEndTablets = new Pair<>(startTablet.map(TabletMetadata::getExtent).orElse(null),
+ endTablet.map(TabletMetadata::getExtent).orElse(null));
+
+ for (TabletMetadata tabletMetadata : tabletMetadatas.values()) {
+ final KeyExtent keyExtent = tabletMetadata.getExtent();
+
+ // Check if this tablet needs to have its files fenced for the deletion
+ if (needsFencingForDeletion(info, keyExtent)) {
+ Manager.log.debug("Found overlapping keyExtent {} for delete, fencing files.", keyExtent);
+
+ // Create the ranges for fencing the files, this takes the place of
+ // chop compactions and splits
+ final List<Range> ranges = createRangesForDeletion(tabletMetadata, range);
+ Preconditions.checkState(!ranges.isEmpty(),
+ "No ranges found that overlap deletion range.");
+
+ // Go through and fence each of the files that are part of the tablet
+ for (Entry<StoredTabletFile,DataFileValue> entry : tabletMetadata.getFilesMap()
+ .entrySet()) {
+ final StoredTabletFile existing = entry.getKey();
+ final DataFileValue value = entry.getValue();
+
+ final Mutation m = new Mutation(keyExtent.toMetaRow());
+
+ // Go through each range that was created and modify the metadata for the file
+ // The end row should be inclusive for the current tablet and the previous end row
+ // should be exclusive for the start row.
+ final Set<StoredTabletFile> newFiles = new HashSet<>();
+ final Set<StoredTabletFile> existingFile = Set.of(existing);
+
+ for (Range fenced : ranges) {
+ // Clip range with the tablet range if the range already exists
+ fenced = existing.hasRange() ? existing.getRange().clip(fenced, true) : fenced;
+
+ // If null the range is disjoint which can happen if there are existing fenced files
+ // If the existing file is disjoint then later we will delete if the file is not part
+ // of the newFiles set which means it is disjoint with all ranges
+ if (fenced != null) {
+ final StoredTabletFile newFile = StoredTabletFile.of(existing.getPath(), fenced);
+ Manager.log.trace("Adding new file {} with range {}", newFile.getMetadataPath(),
+ newFile.getRange());
+
+ // Add the new file to the newFiles set, it will be added later if it doesn't match
+ // the existing file already. We still need to add to the set to be checked later
+ // even if it matches the existing file as later the deletion logic will check to
+ // see if the existing file is part of this set before deleting. This is done to
+ // make sure the existing file isn't deleted unless it is not needed/disjoint
+ // with all ranges.
+ newFiles.add(newFile);
+ } else {
+ Manager.log.trace("Found a disjoint file {} with range {} on delete",
+ existing.getMetadataPath(), existing.getRange());
+ }
+ }
+
+ // If the existingFile is not contained in the newFiles set then we can delete it
+ Sets.difference(existingFile, newFiles).forEach(
+ delete -> m.putDelete(DataFileColumnFamily.NAME, existing.getMetadataText()));
+
+ // Add any new files that don't match the existingFile
+ // As of now we will only have at most 2 files as up to 2 ranges are created
+ final List<StoredTabletFile> filesToAdd =
+ new ArrayList<>(Sets.difference(newFiles, existingFile));
+ Preconditions.checkArgument(filesToAdd.size() <= 2,
+ "There should only be at most 2 StoredTabletFiles after computing new ranges.");
+
+ // If more than 1 new file then re-calculate the num entries and size
+ if (filesToAdd.size() == 2) {
+ // This splits up the values in half and makes sure they total the original
+ // values
+ final Pair<DataFileValue,DataFileValue> newDfvs = computeNewDfv(value);
+ m.put(DataFileColumnFamily.NAME, filesToAdd.get(0).getMetadataText(),
+ newDfvs.getFirst().encodeAsValue());
+ m.put(DataFileColumnFamily.NAME, filesToAdd.get(1).getMetadataText(),
+ newDfvs.getSecond().encodeAsValue());
+ } else {
+ // Will be 0 or 1 files
+ filesToAdd.forEach(newFile -> m.put(DataFileColumnFamily.NAME,
+ newFile.getMetadataText(), value.encodeAsValue()));
+ }
+
+ if (!m.getUpdates().isEmpty()) {
+ bw.addMutation(m);
+ }
+ }
+ } else {
+ Manager.log.debug(
+ "Skipping metadata update on file for keyExtent {} for delete as not overlapping on rows.",
+ keyExtent);
+ }
+ }
+
bw.flush();
+ return startAndEndTablets;
} catch (Exception ex) {
throw new AccumuloException(ex);
}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
index 8eeb7da704,0000000000..7f0be24740
mode 100644,000000..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
@@@ -1,1276 -1,0 +1,1276 @@@
+/*
+ * 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.compaction.coordinator;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.COMPACTED;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.ECOMP;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.FILES;
+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.FileNotFoundException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+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;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.TableDeletedException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.CompactionConfig;
+import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
+import org.apache.accumulo.core.clientImpl.thrift.TInfo;
+import org.apache.accumulo.core.clientImpl.thrift.TableOperation;
+import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType;
+import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
+import org.apache.accumulo.core.clientImpl.thrift.ThriftTableOperationException;
+import org.apache.accumulo.core.compaction.thrift.CompactionCoordinatorService;
+import org.apache.accumulo.core.compaction.thrift.TCompactionState;
+import org.apache.accumulo.core.compaction.thrift.TCompactionStatusUpdate;
+import org.apache.accumulo.core.compaction.thrift.TExternalCompaction;
+import org.apache.accumulo.core.compaction.thrift.TExternalCompactionList;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.NamespaceId;
+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.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.iteratorsImpl.system.SystemIteratorUtil;
+import org.apache.accumulo.core.metadata.AbstractTabletFile;
+import org.apache.accumulo.core.metadata.CompactableFileImpl;
+import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+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;
+import org.apache.accumulo.core.metadata.schema.SelectedFiles;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.rpc.ThriftUtil;
+import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
+import org.apache.accumulo.core.spi.compaction.CompactionJob;
+import org.apache.accumulo.core.spi.compaction.CompactionKind;
+import org.apache.accumulo.core.tabletserver.thrift.InputFile;
+import org.apache.accumulo.core.tabletserver.thrift.IteratorConfig;
+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.TabletServerClientService;
+import org.apache.accumulo.core.util.Retry;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.core.util.cache.Caches.CacheName;
+import org.apache.accumulo.core.util.compaction.CompactionExecutorIdImpl;
+import org.apache.accumulo.core.util.compaction.ExternalCompactionUtil;
+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;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.security.SecurityOperation;
+import org.apache.accumulo.server.tablets.TabletNameGenerator;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.thrift.TException;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.CacheLoader;
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.github.benmanes.caffeine.cache.Weigher;
+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 {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CompactionCoordinator.class);
+ private static final long FIFTEEN_MINUTES = TimeUnit.MINUTES.toMillis(15);
+
+ /*
+ * Map of compactionId to RunningCompactions. This is an informational cache of what external
+ * compactions may be running. Its possible it may contain external compactions that are not
+ * actually running. It may not contain compactions that are actually running. The metadata table
+ * is the most authoritative source of what external compactions are currently running, but it
+ * does not have the stats that this map has.
+ */
+ protected static final Map<ExternalCompactionId,RunningCompaction> 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<Ample.DataLevel,CountDownLatch> refreshLatches;
+
+ /* Map of group name to last time compactor called to get a compaction job */
+ // ELASTICITY_TODO need to clean out groups that are no longer configured..
+ private static final Map<String,Long> TIME_COMPACTOR_LAST_CHECKED = new ConcurrentHashMap<>();
+
+ private final ServerContext ctx;
+ private final LiveTServerSet tserverSet;
+ private final SecurityOperation security;
+ private final CompactionJobQueues jobQueues;
+ // Exposed for tests
+ protected volatile Boolean shutdown = false;
+
+ private final ScheduledThreadPoolExecutor schedExecutor;
+
+ private final Cache<ExternalCompactionId,RunningCompaction> completed;
+ private LoadingCache<Long,CompactionConfig> compactionConfigCache;
+ private final Cache<Path,Integer> checked_tablet_dir_cache;
+
+ public CompactionCoordinator(ServerContext ctx, LiveTServerSet tservers,
+ SecurityOperation security, CompactionJobQueues jobQueues) {
+ this.ctx = ctx;
+ this.tserverSet = tservers;
+ this.schedExecutor = this.ctx.getScheduledExecutor();
+ this.security = security;
+ this.jobQueues = jobQueues;
+
+ var refreshLatches = new EnumMap<Ample.DataLevel,CountDownLatch>(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);
+
+ completed = ctx.getCaches().createNewBuilder(CacheName.COMPACTIONS_COMPLETED, true)
+ .maximumSize(200).expireAfterWrite(10, TimeUnit.MINUTES).build();
+
+ CacheLoader<Long,CompactionConfig> loader =
+ txid -> CompactionConfigStorage.getConfig(ctx, txid);
+
+ // Keep a small short lived cache of compaction config. Compaction config never changes, however
+ // when a compaction is canceled it is deleted which is why there is a time limit. It does not
+ // hurt to let a job that was canceled start, it will be canceled later. Caching this immutable
+ // config will help avoid reading the same data over and over.
+ compactionConfigCache = ctx.getCaches().createNewBuilder(CacheName.COMPACTION_CONFIGS, true)
+ .expireAfterWrite(30, SECONDS).maximumSize(100).build(loader);
+
+ Weigher<Path,Integer> weigher = (path, count) -> {
+ return path.toUri().toString().length();
+ };
+
+ checked_tablet_dir_cache =
+ ctx.getCaches().createNewBuilder(CacheName.COMPACTION_DIR_CACHE, true)
+ .maximumWeight(10485760L).weigher(weigher).build();
+
+ // At this point the manager does not have its lock so no actions should be taken yet
+ }
+
+ public void shutdown() {
+ shutdown = true;
+ }
+
+ protected void startCompactionCleaner(ScheduledThreadPoolExecutor schedExecutor) {
+ ScheduledFuture<?> future =
+ schedExecutor.scheduleWithFixedDelay(this::cleanUpCompactors, 0, 5, TimeUnit.MINUTES);
+ ThreadPools.watchNonCriticalScheduledTask(future);
+ }
+
+ protected void startRunningCleaner(ScheduledThreadPoolExecutor schedExecutor) {
+ ScheduledFuture<?> future =
+ schedExecutor.scheduleWithFixedDelay(this::cleanUpRunning, 0, 5, TimeUnit.MINUTES);
+ ThreadPools.watchNonCriticalScheduledTask(future);
+ }
+
+ private void processRefreshes(Ample.DataLevel dataLevel) {
+ try (var refreshStream = ctx.getAmple().refreshes(dataLevel).stream()) {
+ // process batches of refresh entries to avoid reading all into memory at once
+ Iterators.partition(refreshStream.iterator(), 10000).forEachRemaining(refreshEntries -> {
+ LOG.info("Processing {} tablet refreshes for {}", refreshEntries.size(), dataLevel);
+
+ var extents =
+ refreshEntries.stream().map(RefreshEntry::getExtent).collect(Collectors.toList());
+ var tabletsMeta = new HashMap<KeyExtent,TabletMetadata>();
+ try (var tablets = ctx.getAmple().readTablets().forTablets(extents, Optional.empty())
+ .fetch(PREV_ROW, LOCATION, SCANS).build()) {
+ tablets.stream().forEach(tm -> tabletsMeta.put(tm.getExtent(), tm));
+ }
+
+ var tserverRefreshes = new HashMap<TabletMetadata.Location,List<TKeyExtent>>();
+
+ 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<StoredTabletFile> scanfiles = tm.getScans();
+ var ttr = extent.toThrift();
+ 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.
+ LOG.info("Checking for running external compactions");
+ // On re-start contact the running Compactors to try and seed the list of running compactions
+ List<RunningCompaction> running = getCompactionsRunningOnCompactors();
+ if (running.isEmpty()) {
+ LOG.info("No running external compactions found");
+ } else {
+ LOG.info("Found {} running external compactions", running.size());
+ running.forEach(rc -> {
+ TCompactionStatusUpdate update = new TCompactionStatusUpdate();
+ update.setState(TCompactionState.IN_PROGRESS);
+ update.setMessage("Coordinator restarted, compaction found in progress");
+ rc.addUpdate(System.currentTimeMillis(), update);
+ RUNNING_CACHE.put(ExternalCompactionId.of(rc.getJob().getExternalCompactionId()), rc);
+ });
+ }
+
+ startDeadCompactionDetector();
+
+ // ELASTICITY_TODO the main function of the following loop was getting group summaries from
+ // tservers. Its no longer doing that. May be best to remove the loop and make the remaining
+ // task a scheduled one.
+
+ LOG.info("Starting loop to check tservers for compaction summaries");
+ while (!shutdown) {
+ long start = System.currentTimeMillis();
+
+ long now = System.currentTimeMillis();
+ TIME_COMPACTOR_LAST_CHECKED.forEach((k, v) -> {
+ if ((now - v) > getMissingCompactorWarningTime()) {
+ // ELASTICITY_TODO may want to consider of the group has any jobs queued OR if the group
+ // still exist in configuration
+ LOG.warn("No compactors have checked in with coordinator for group {} in {}ms", k,
+ getMissingCompactorWarningTime());
+ }
+ });
+
+ long checkInterval = getTServerCheckInterval();
+ long duration = (System.currentTimeMillis() - start);
+ if (checkInterval - duration > 0) {
+ LOG.debug("Waiting {}ms for next group check", (checkInterval - duration));
+ UtilWaitThread.sleep(checkInterval - duration);
+ }
+ }
+
+ LOG.info("Shutting down");
+ }
+
+ protected void startDeadCompactionDetector() {
+ new DeadCompactionDetector(this.ctx, this, schedExecutor).start();
+ }
+
+ protected long getMissingCompactorWarningTime() {
+ return FIFTEEN_MINUTES;
+ }
+
+ protected long getTServerCheckInterval() {
+ return this.ctx.getConfiguration()
+ .getTimeInMillis(Property.COMPACTION_COORDINATOR_TSERVER_COMPACTION_CHECK_INTERVAL);
+ }
+
+ /**
+ * Callback for the LiveTServerSet object to update current set of tablet servers, including ones
+ * that were deleted and added
+ *
+ * @param current current set of live tservers
+ * @param deleted set of tservers that were removed from current since last update
+ * @param added set of tservers that were added to current since last update
+ */
+ public void updateTServerSet(LiveTServerSet current, Set<TServerInstance> deleted,
+ Set<TServerInstance> added) {
+
+ }
+
+ /**
+ * Return the next compaction job from the queue to a Compactor
+ *
+ * @param groupName group
+ * @param compactorAddress compactor address
+ * @throws ThriftSecurityException when permission error
+ * @return compaction job
+ */
+ @Override
+ public TExternalCompactionJob getCompactionJob(TInfo tinfo, TCredentials credentials,
+ String groupName, String compactorAddress, String externalCompactionId)
+ throws ThriftSecurityException {
+
+ // do not expect users to call this directly, expect compactors to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ final String group = groupName.intern();
+ LOG.trace("getCompactionJob called for group {} by compactor {}", group, compactorAddress);
+ TIME_COMPACTOR_LAST_CHECKED.put(group, System.currentTimeMillis());
+
+ TExternalCompactionJob result = null;
+
+ CompactionJobQueues.MetaJob metaJob =
+ jobQueues.poll(CompactionExecutorIdImpl.externalId(groupName));
+
+ while (metaJob != null) {
+
+ Optional<CompactionConfig> compactionConfig = getCompactionConfig(metaJob);
+
+ // this method may reread the metadata, do not use the metadata in metaJob for anything after
+ // this method
+ ExternalCompactionMetadata ecm = null;
+
+ var kind = metaJob.getJob().getKind();
+
+ // Only reserve user compactions when the config is present. When compactions are canceled the
+ // config is deleted.
+ if (kind == CompactionKind.SYSTEM
+ || (kind == CompactionKind.USER && compactionConfig.isPresent())) {
+ ecm = reserveCompaction(metaJob, compactorAddress, externalCompactionId);
+ }
+
+ if (ecm != null) {
+ result = createThriftJob(externalCompactionId, ecm, metaJob, compactionConfig);
+ // It is possible that by the time this added that the the compactor that made this request
+ // is dead. In this cases the compaction is not actually running.
+ RUNNING_CACHE.put(ExternalCompactionId.of(result.getExternalCompactionId()),
+ new RunningCompaction(result, compactorAddress, group));
+ LOG.debug("Returning external job {} to {} with {} files", result.externalCompactionId,
+ compactorAddress, ecm.getJobFiles().size());
+ break;
+ } else {
+ LOG.debug("Unable to reserve compaction job for {}, pulling another off the queue ",
+ metaJob.getTabletMetadata().getExtent());
+ metaJob = jobQueues.poll(CompactionExecutorIdImpl.externalId(groupName));
+ }
+ }
+
+ if (metaJob == null) {
+ LOG.debug("No jobs found in group {} ", group);
+ }
+
+ if (result == null) {
+ LOG.trace("No jobs found for group {}, returning empty job to compactor {}", group,
+ compactorAddress);
+ result = new TExternalCompactionJob();
+ }
+
+ return result;
+
+ }
+
+ // ELASTICITY_TODO unit test this code
+ private boolean canReserveCompaction(TabletMetadata tablet, CompactionJob job,
+ Set<StoredTabletFile> jobFiles) {
+
+ if (tablet == null) {
+ // the tablet no longer exist
+ return false;
+ }
+
+ if (tablet.getOperationId() != null) {
+ return false;
+ }
+
+ if (!tablet.getFiles().containsAll(jobFiles)) {
+ return false;
+ }
+
+ var currentlyCompactingFiles = tablet.getExternalCompactions().values().stream()
+ .flatMap(ecm -> ecm.getJobFiles().stream()).collect(Collectors.toSet());
+
+ if (!Collections.disjoint(jobFiles, currentlyCompactingFiles)) {
+ return false;
+ }
+
+ switch (job.getKind()) {
+ case SYSTEM:
+ if (tablet.getSelectedFiles() != null
+ && !Collections.disjoint(jobFiles, tablet.getSelectedFiles().getFiles())) {
+ return false;
+ }
+ break;
+ case USER:
+ case SELECTOR:
+ if (tablet.getSelectedFiles() == null
+ || !tablet.getSelectedFiles().getFiles().containsAll(jobFiles)) {
+ return false;
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException("Not currently handling " + job.getKind());
+ }
+
+ return true;
+ }
+
+ private void checkTabletDir(KeyExtent extent, Path path) {
+ try {
+ if (checked_tablet_dir_cache.getIfPresent(path) == null) {
+ FileStatus[] files = null;
+ try {
+ files = ctx.getVolumeManager().listStatus(path);
+ } catch (FileNotFoundException ex) {
+ // ignored
+ }
+
+ if (files == null) {
+ LOG.debug("Tablet {} had no dir, creating {}", extent, path);
+
+ ctx.getVolumeManager().mkdirs(path);
+ }
+ checked_tablet_dir_cache.put(path, 1);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private ExternalCompactionMetadata createExternalCompactionMetadata(CompactionJob job,
+ Set<StoredTabletFile> jobFiles, TabletMetadata tablet, String compactorAddress) {
+ boolean propDels;
+
+ Long fateTxId = null;
+
+ switch (job.getKind()) {
+ case SYSTEM: {
+ boolean compactingAll = tablet.getFiles().equals(jobFiles);
+ propDels = !compactingAll;
+ }
+ break;
+ case SELECTOR:
+ case USER: {
+ boolean compactingAll = tablet.getSelectedFiles().initiallySelectedAll()
+ && tablet.getSelectedFiles().getFiles().equals(jobFiles);
+ propDels = !compactingAll;
+ fateTxId = tablet.getSelectedFiles().getFateTxId();
+ }
+ break;
+ default:
+ throw new IllegalArgumentException();
+ }
+
+ Consumer<String> directoryCreator = dir -> checkTabletDir(tablet.getExtent(), new Path(dir));
+ ReferencedTabletFile newFile =
+ TabletNameGenerator.getNextDataFilenameForMajc(propDels, ctx, tablet, directoryCreator);
+
+ return new ExternalCompactionMetadata(jobFiles, newFile, compactorAddress, job.getKind(),
+ job.getPriority(), job.getExecutor(), propDels, fateTxId);
+
+ }
+
+ private ExternalCompactionMetadata reserveCompaction(CompactionJobQueues.MetaJob metaJob,
+ String compactorAddress, String externalCompactionId) {
+
+ Preconditions.checkArgument(metaJob.getJob().getKind() == CompactionKind.SYSTEM
+ || metaJob.getJob().getKind() == CompactionKind.USER);
+
+ var tabletMetadata = metaJob.getTabletMetadata();
+
+ var jobFiles = metaJob.getJob().getFiles().stream().map(CompactableFileImpl::toStoredTabletFile)
+ .collect(Collectors.toSet());
+
+ Retry retry =
+ Retry.builder().maxRetries(5).retryAfter(100, MILLISECONDS).incrementBy(100, MILLISECONDS)
+ .maxWait(10, SECONDS).backOffFactor(1.5).logInterval(3, MINUTES).createRetry();
+
+ while (retry.canRetry()) {
+ try (var tabletsMutator = ctx.getAmple().conditionallyMutateTablets()) {
+ var extent = metaJob.getTabletMetadata().getExtent();
+
+ if (!canReserveCompaction(tabletMetadata, metaJob.getJob(), jobFiles)) {
+ return null;
+ }
+
+ var ecm = createExternalCompactionMetadata(metaJob.getJob(), jobFiles, tabletMetadata,
+ compactorAddress);
+
+ // any data that is read from the tablet to make a decision about if it can compact or not
+ // must be included in the requireSame call
+ var tabletMutator = tabletsMutator.mutateTablet(extent).requireAbsentOperation()
+ .requireSame(tabletMetadata, PREV_ROW, FILES, SELECTED, ECOMP);
+
+ var ecid = ExternalCompactionId.of(externalCompactionId);
+ tabletMutator.putExternalCompaction(ecid, ecm);
+ tabletMutator.submit(tm -> tm.getExternalCompactions().containsKey(ecid));
+
+ var result = tabletsMutator.process().get(extent);
+
+ if (result.getStatus() == Ample.ConditionalResult.Status.ACCEPTED) {
+ return ecm;
+ } else {
+ tabletMetadata = result.readMetadata();
+ }
+ }
+
+ retry.useRetry();
+ try {
+ retry.waitForNextAttempt(LOG,
+ "Reserved compaction for " + metaJob.getTabletMetadata().getExtent());
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ return null;
+ }
+
+ TExternalCompactionJob createThriftJob(String externalCompactionId,
+ ExternalCompactionMetadata ecm, CompactionJobQueues.MetaJob metaJob,
+ Optional<CompactionConfig> compactionConfig) {
+
+ Map<String,String> overrides = CompactionPluginUtils.computeOverrides(compactionConfig, ctx,
+ metaJob.getTabletMetadata().getExtent(), metaJob.getJob().getFiles());
+
+ IteratorConfig iteratorSettings = SystemIteratorUtil
+ .toIteratorConfig(compactionConfig.map(CompactionConfig::getIterators).orElse(List.of()));
+
+ var files = ecm.getJobFiles().stream().map(storedTabletFile -> {
+ var dfv = metaJob.getTabletMetadata().getFilesMap().get(storedTabletFile);
- return new InputFile(storedTabletFile.getNormalizedPathStr(), dfv.getSize(),
- dfv.getNumEntries(), dfv.getTime());
++ return new InputFile(storedTabletFile.getMetadata(), dfv.getSize(), dfv.getNumEntries(),
++ dfv.getTime());
+ }).collect(Collectors.toList());
+
+ long fateTxid = 0;
+ if (metaJob.getJob().getKind() == CompactionKind.USER) {
+ fateTxid = metaJob.getTabletMetadata().getSelectedFiles().getFateTxId();
+ }
+
+ return new TExternalCompactionJob(externalCompactionId,
+ metaJob.getTabletMetadata().getExtent().toThrift(), files, iteratorSettings,
+ ecm.getCompactTmpName().getNormalizedPathStr(), ecm.getPropagateDeletes(),
+ TCompactionKind.valueOf(ecm.getKind().name()), fateTxid, 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<CompactionConfig> getCompactionConfig(CompactionJobQueues.MetaJob metaJob) {
+ if (metaJob.getJob().getKind() == CompactionKind.USER
+ && metaJob.getTabletMetadata().getSelectedFiles() != null) {
+ var cconf =
+ compactionConfigCache.get(metaJob.getTabletMetadata().getSelectedFiles().getFateTxId());
+ return Optional.ofNullable(cconf);
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Compactors calls this method when they have finished a compaction. This method does the
+ * following.
+ *
+ * <ol>
+ * <li>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.</li>
+ * <li>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.</li>
+ * <li>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.</li>
+ * <li>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.</li>
+ * <li>If a ~refresh entry was written, delete it since the refresh was successful.</li>
+ * </ol>
+ *
+ * <p>
+ * 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.
+ * </p>
+ *
+ * <p>
+ * 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.
+ * </p>
+ *
+ * <p>
+ * Issue #3559 was opened to explore the possibility of making compaction commit a fate operation
+ * which would remove the need for the ~refresh section.
+ * </p>
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @param externalCompactionId compaction id
+ * @param textent tablet extent
+ * @param stats compaction stats
+ * @throws ThriftSecurityException when permission error
+ */
+ @Override
+ public void compactionCompleted(TInfo tinfo, TCredentials credentials,
+ String externalCompactionId, TKeyExtent textent, TCompactionStats stats)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+
+ var extent = KeyExtent.fromThrift(textent);
+ LOG.info("Compaction completed, id: {}, stats: {}, extent: {}", externalCompactionId, stats,
+ extent);
+ final var ecid = ExternalCompactionId.of(externalCompactionId);
+
+ var tabletMeta =
+ ctx.getAmple().readTablet(extent, ECOMP, SELECTED, LOCATION, FILES, COMPACTED, OPID);
+
+ if (!canCommitCompaction(ecid, tabletMeta)) {
+ return;
+ }
+
+ ExternalCompactionMetadata ecm = tabletMeta.getExternalCompactions().get(ecid);
+
+ // ELASTICITY_TODO this code does not handle race conditions or faults. Need to ensure refresh
+ // happens in the case of manager process death between commit and refresh.
+ ReferencedTabletFile newDatafile =
+ TabletNameGenerator.computeCompactionFileDest(ecm.getCompactTmpName());
+
+ Optional<ReferencedTabletFile> optionalNewFile;
+ try {
+ optionalNewFile = renameOrDeleteFile(stats, ecm, newDatafile);
+ } catch (IOException e) {
+ LOG.warn("Can not commit complete compaction {} because unable to delete or rename {} ", ecid,
+ ecm.getCompactTmpName(), e);
+ compactionFailed(Map.of(ecid, extent));
+ return;
+ }
+
+ RefreshWriter refreshWriter = new RefreshWriter(ecid, extent);
+
+ try {
+ 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));
+ }
+
+ if (ecm.getKind() != CompactionKind.USER) {
+ refreshTablet(tabletMeta);
+ }
+
+ // 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
+ // corresponding external compaction.
+ recordCompletion(ecid);
+ }
+
+ private Optional<ReferencedTabletFile> renameOrDeleteFile(TCompactionStats stats,
+ ExternalCompactionMetadata ecm, ReferencedTabletFile newDatafile) throws IOException {
+ if (stats.getEntriesWritten() == 0) {
+ // the compaction produced no output so do not need to rename or add a file to the metadata
+ // table, only delete the input files.
+ if (!ctx.getVolumeManager().delete(ecm.getCompactTmpName().getPath())) {
+ throw new IOException("delete returned false");
+ }
+
+ return Optional.empty();
+ } else {
+ if (!ctx.getVolumeManager().rename(ecm.getCompactTmpName().getPath(),
+ newDatafile.getPath())) {
+ throw new IOException("rename returned false");
+ }
+
+ return Optional.of(newDatafile);
+ }
+ }
+
+ private void refreshTablet(TabletMetadata metadata) {
+ var location = metadata.getLocation();
+ if (location != null) {
+ KeyExtent extent = metadata.getExtent();
+
+ // 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 {
+ TabletRefresher.refreshTablets(executorService,
+ "compaction:" + metadata.getExtent().toString(), ctx, tserverSet::getCurrentServers,
+ Map.of(metadata.getLocation(), List.of(extent.toThrift())));
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+ }
+
+ // ELASTICITY_TODO unit test this method
+ private boolean canCommitCompaction(ExternalCompactionId ecid, TabletMetadata tabletMetadata) {
+
+ if (tabletMetadata == null) {
+ LOG.debug("Received completion notification for nonexistent tablet {}", ecid);
+ return false;
+ }
+
+ var extent = tabletMetadata.getExtent();
+
+ if (tabletMetadata.getOperationId() != null) {
+ // split, merge, and delete tablet should delete the compaction entry in the tablet
+ LOG.debug("Received completion notification for tablet with active operation {} {} {}", ecid,
+ extent, tabletMetadata.getOperationId());
+ return false;
+ }
+
+ ExternalCompactionMetadata ecm = tabletMetadata.getExternalCompactions().get(ecid);
+
+ if (ecm == null) {
+ LOG.debug("Received completion notification for unknown compaction {} {}", ecid, extent);
+ return false;
+ }
+
+ if (ecm.getKind() == CompactionKind.USER || ecm.getKind() == CompactionKind.SELECTOR) {
+ if (tabletMetadata.getSelectedFiles() == null) {
+ // when the compaction is canceled, selected files are deleted
+ LOG.debug(
+ "Received completion notification for user compaction and tablet has no selected files {} {}",
+ ecid, extent);
+ return false;
+ }
+
+ if (ecm.getFateTxId() != tabletMetadata.getSelectedFiles().getFateTxId()) {
+ // maybe the compaction was cancled and another user compaction was started on the tablet.
+ LOG.debug(
+ "Received completion notification for user compaction where its fate txid did not match the tablets {} {} {} {}",
+ ecid, extent, FateTxId.formatTid(ecm.getFateTxId()),
+ FateTxId.formatTid(tabletMetadata.getSelectedFiles().getFateTxId()));
+ }
+
+ if (!tabletMetadata.getSelectedFiles().getFiles().containsAll(ecm.getJobFiles())) {
+ // this is not expected to happen
+ LOG.error("User compaction contained files not in the selected set {} {} {} {} {}",
+ tabletMetadata.getExtent(), ecid, ecm.getKind(),
+ Optional.ofNullable(tabletMetadata.getSelectedFiles()).map(SelectedFiles::getFiles),
+ ecm.getJobFiles());
+ return false;
+ }
+ }
+
+ if (!tabletMetadata.getFiles().containsAll(ecm.getJobFiles())) {
+ // this is not expected to happen
+ LOG.error("Compaction contained files not in the tablet files set {} {} {} {}",
+ tabletMetadata.getExtent(), ecid, tabletMetadata.getFiles(), ecm.getJobFiles());
+ return false;
+ }
+
+ return true;
+ }
+
+ private TabletMetadata commitCompaction(TCompactionStats stats, ExternalCompactionId ecid,
+ TabletMetadata tablet, Optional<ReferencedTabletFile> newDatafile,
+ RefreshWriter refreshWriter) {
+
+ KeyExtent extent = tablet.getExtent();
+
+ Retry retry = Retry.builder().infiniteRetries().retryAfter(100, MILLISECONDS)
+ .incrementBy(100, MILLISECONDS).maxWait(10, SECONDS).backOffFactor(1.5)
+ .logInterval(3, MINUTES).createRetry();
+
+ while (canCommitCompaction(ecid, tablet)) {
+ ExternalCompactionMetadata ecm = tablet.getExternalCompactions().get(ecid);
+
+ // the compacted files should not exists in the tablet already
+ var tablet2 = tablet;
+ newDatafile.ifPresent(
+ newFile -> Preconditions.checkState(!tablet2.getFiles().contains(newFile.insert()),
+ "File already exists in tablet %s %s", newFile, tablet2.getFiles()));
+
+ 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, LOCATION);
+
+ if (ecm.getKind() == CompactionKind.USER || ecm.getKind() == CompactionKind.SELECTOR) {
+ tabletMutator.requireSame(tablet, SELECTED, COMPACTED);
+ }
+
+ // make the needed updates to the tablet
+ updateTabletForCompaction(stats, ecid, tablet, newDatafile, extent, ecm, tabletMutator);
+
+ tabletMutator
+ .submit(tabletMetadata -> !tabletMetadata.getExternalCompactions().containsKey(ecid));
+
+ // TODO expensive logging
+ LOG.debug("Compaction completed {} added {} removed {}", tablet.getExtent(), newDatafile,
+ ecm.getJobFiles().stream().map(AbstractTabletFile::getFileName)
+ .collect(Collectors.toList()));
+
+ // ELASTICITY_TODO check return value and retry, could fail because of race conditions
+ var result = tabletsMutator.process().get(extent);
+ if (result.getStatus() == Ample.ConditionalResult.Status.ACCEPTED) {
+ // compaction was committed
+ break;
+ } else {
+ // compaction failed to commit, maybe something changed on the tablet so lets reread the
+ // metadata and try again
+ tablet = result.readMetadata();
+ }
+
+ retry.waitForNextAttempt(LOG, "Failed to commit " + ecid + " for tablet " + extent);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ return tablet;
+ }
+
+ private void updateTabletForCompaction(TCompactionStats stats, ExternalCompactionId ecid,
+ TabletMetadata tablet, Optional<ReferencedTabletFile> newDatafile, KeyExtent extent,
+ ExternalCompactionMetadata ecm, Ample.ConditionalTabletMutator tabletMutator) {
+ // ELASTICITY_TODO improve logging adapt to use existing tablet files logging
+ if (ecm.getKind() == CompactionKind.USER) {
+ if (tablet.getSelectedFiles().getFiles().equals(ecm.getJobFiles())) {
+ // all files selected for the user compactions are finished, so the tablet is finish and
+ // its compaction id needs to be updated.
+
+ long fateTxId = tablet.getSelectedFiles().getFateTxId();
+
+ Preconditions.checkArgument(!tablet.getCompacted().contains(fateTxId),
+ "Tablet %s unexpected has selected files and compacted columns for %s",
+ tablet.getExtent(), fateTxId);
+
+ // TODO set to trace
+ LOG.debug("All selected files compcated for {} setting compacted for {}",
+ tablet.getExtent(), FateTxId.formatTid(tablet.getSelectedFiles().getFateTxId()));
+
+ tabletMutator.deleteSelectedFiles();
+ tabletMutator.putCompacted(fateTxId);
+
+ } else {
+ // not all of the selected files were finished, so need to add the new file to the
+ // selected set
+
+ Set<StoredTabletFile> newSelectedFileSet =
+ new HashSet<>(tablet.getSelectedFiles().getFiles());
+ newSelectedFileSet.removeAll(ecm.getJobFiles());
+
+ if (newDatafile.isPresent()) {
+ // TODO set to trace
+ LOG.debug(
+ "Not all selected files for {} are done, adding new selected file {} from compaction",
+ tablet.getExtent(), newDatafile.orElseThrow().getPath().getName());
+ newSelectedFileSet.add(newDatafile.orElseThrow().insert());
+ } else {
+ // TODO set to trace
+ LOG.debug(
+ "Not all selected files for {} are done, compaction produced no output so not adding to selected set.",
+ tablet.getExtent());
+ }
+
+ tabletMutator.putSelectedFiles(
+ new SelectedFiles(newSelectedFileSet, tablet.getSelectedFiles().initiallySelectedAll(),
+ tablet.getSelectedFiles().getFateTxId()));
+ }
+ }
+
+ 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);
+
+ if (newDatafile.isPresent()) {
+ tabletMutator.putFile(newDatafile.orElseThrow(),
+ new DataFileValue(stats.getFileSize(), stats.getEntriesWritten()));
+ }
+ }
+
+ @Override
+ public void compactionFailed(TInfo tinfo, TCredentials credentials, String externalCompactionId,
+ TKeyExtent extent) throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ LOG.info("Compaction failed, id: {}", externalCompactionId);
+ final var ecid = ExternalCompactionId.of(externalCompactionId);
+ compactionFailed(Map.of(ecid, KeyExtent.fromThrift(extent)));
+
+ // ELASTICITIY_TODO need to open an issue about making the GC clean up tmp files. The tablet
+ // currently cleans up tmp files on tablet load. With tablets never loading possibly but still
+ // compacting dying compactors may still leave tmp files behind.
+ }
+
+ void compactionFailed(Map<ExternalCompactionId,KeyExtent> compactions) {
+
+ try (var tabletsMutator = ctx.getAmple().conditionallyMutateTablets()) {
+ compactions.forEach((ecid, extent) -> {
+ try {
+ ctx.requireNotDeleted(extent.tableId());
+ tabletsMutator.mutateTablet(extent).requireAbsentOperation().requireCompaction(ecid)
+ .requirePrevEndRow(extent.prevEndRow()).deleteExternalCompaction(ecid)
+ .submit(tabletMetadata -> !tabletMetadata.getExternalCompactions().containsKey(ecid));
+ } catch (TableDeletedException e) {
+ LOG.warn("Table {} was deleted, unable to update metadata for compaction failure.",
+ extent.tableId());
+ }
+ });
+
+ tabletsMutator.process().forEach((extent, result) -> {
+ if (result.getStatus() != Ample.ConditionalResult.Status.ACCEPTED) {
+ // this should try again later when the dead compaction detector runs, lets log it in case
+ // its a persistent problem
+ if (LOG.isDebugEnabled()) {
+ var ecid =
+ compactions.entrySet().stream().filter(entry -> entry.getValue().equals(extent))
+ .findFirst().map(Map.Entry::getKey).orElse(null);
+ LOG.debug("Unable to remove failed compaction {} {}", extent, ecid);
+ }
+ }
+ });
+ }
+
+ compactions.forEach((k, v) -> recordCompletion(k));
+ }
+
+ /**
+ * Compactor calls to update the status of the assigned compaction
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @param externalCompactionId compaction id
+ * @param update compaction status update
+ * @param timestamp timestamp of the message
+ * @throws ThriftSecurityException when permission error
+ */
+ @Override
+ public void updateCompactionStatus(TInfo tinfo, TCredentials credentials,
+ String externalCompactionId, TCompactionStatusUpdate update, long timestamp)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ LOG.debug("Compaction status update, id: {}, timestamp: {}, update: {}", externalCompactionId,
+ timestamp, update);
+ final RunningCompaction rc = RUNNING_CACHE.get(ExternalCompactionId.of(externalCompactionId));
+ if (null != rc) {
+ rc.addUpdate(timestamp, update);
+ }
+ }
+
+ private void recordCompletion(ExternalCompactionId ecid) {
+ var rc = RUNNING_CACHE.remove(ecid);
+ if (rc != null) {
+ completed.put(ecid, rc);
+ }
+ }
+
+ protected Set<ExternalCompactionId> readExternalCompactionIds() {
+ return this.ctx.getAmple().readTablets().forLevel(Ample.DataLevel.USER).fetch(ECOMP).build()
+ .stream().flatMap(tm -> tm.getExternalCompactions().keySet().stream())
+ .collect(Collectors.toSet());
+ }
+
+ /**
+ * The RUNNING_CACHE set may contain external compactions that are not actually running. This
+ * method periodically cleans those up.
+ */
+ protected void cleanUpRunning() {
+
+ // grab a snapshot of the ids in the set before reading the metadata table. This is done to
+ // avoid removing things that are added while reading the metadata.
+ Set<ExternalCompactionId> idsSnapshot = Set.copyOf(RUNNING_CACHE.keySet());
+
+ // grab the ids that are listed as running in the metadata table. It important that this is done
+ // after getting the snapshot.
+ Set<ExternalCompactionId> idsInMetadata = readExternalCompactionIds();
+
+ var idsToRemove = Sets.difference(idsSnapshot, idsInMetadata);
+
+ // remove ids that are in the running set but not in the metadata table
+ idsToRemove.forEach(this::recordCompletion);
+
+ if (idsToRemove.size() > 0) {
+ LOG.debug("Removed stale entries from RUNNING_CACHE : {}", idsToRemove);
+ }
+ }
+
+ /**
+ * Return information about running compactions
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @return map of ECID to TExternalCompaction objects
+ * @throws ThriftSecurityException permission error
+ */
+ @Override
+ public TExternalCompactionList getRunningCompactions(TInfo tinfo, TCredentials credentials)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+
+ final TExternalCompactionList result = new TExternalCompactionList();
+ RUNNING_CACHE.forEach((ecid, rc) -> {
+ TExternalCompaction trc = new TExternalCompaction();
+ trc.setGroupName(rc.getGroupName());
+ trc.setCompactor(rc.getCompactorAddress());
+ trc.setUpdates(rc.getUpdates());
+ trc.setJob(rc.getJob());
+ result.putToCompactions(ecid.canonical(), trc);
+ });
+ return result;
+ }
+
+ /**
+ * Return information about recently completed compactions
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @return map of ECID to TExternalCompaction objects
+ * @throws ThriftSecurityException permission error
+ */
+ @Override
+ public TExternalCompactionList getCompletedCompactions(TInfo tinfo, TCredentials credentials)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ final TExternalCompactionList result = new TExternalCompactionList();
+ completed.asMap().forEach((ecid, rc) -> {
+ TExternalCompaction trc = new TExternalCompaction();
+ trc.setGroupName(rc.getGroupName());
+ trc.setCompactor(rc.getCompactorAddress());
+ trc.setJob(rc.getJob());
+ trc.setUpdates(rc.getUpdates());
+ result.putToCompactions(ecid.canonical(), trc);
+ });
+ return result;
+ }
+
+ @Override
+ public void cancel(TInfo tinfo, TCredentials credentials, String externalCompactionId)
+ throws TException {
+ var runningCompaction = RUNNING_CACHE.get(ExternalCompactionId.of(externalCompactionId));
+ var extent = KeyExtent.fromThrift(runningCompaction.getJob().getExtent());
+ try {
+ NamespaceId nsId = this.ctx.getNamespaceId(extent.tableId());
+ if (!security.canCompact(credentials, extent.tableId(), nsId)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ } catch (TableNotFoundException e) {
+ throw new ThriftTableOperationException(extent.tableId().canonical(), null,
+ TableOperation.COMPACT_CANCEL, TableOperationExceptionType.NOTFOUND, e.getMessage());
+ }
+
+ cancelCompactionOnCompactor(runningCompaction.getCompactorAddress(), externalCompactionId);
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected String getTServerAddressString(HostAndPort tserverAddress) {
+ return ExternalCompactionUtil.getHostPortString(tserverAddress);
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected List<RunningCompaction> getCompactionsRunningOnCompactors() {
+ return ExternalCompactionUtil.getCompactionsRunningOnCompactors(this.ctx);
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected void cancelCompactionOnCompactor(String address, String externalCompactionId) {
+ HostAndPort hostPort = HostAndPort.fromString(address);
+ ExternalCompactionUtil.cancelCompaction(this.ctx, hostPort, externalCompactionId);
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected void returnTServerClient(TabletServerClientService.Client client) {
+ ThriftUtil.returnClient(client, this.ctx);
+ }
+
+ private void deleteEmpty(ZooReaderWriter zoorw, String path)
+ throws KeeperException, InterruptedException {
+ try {
+ LOG.debug("Deleting empty ZK node {}", path);
+ zoorw.delete(path);
+ } catch (KeeperException.NotEmptyException e) {
+ LOG.debug("Failed to delete {} its not empty, likely an expected race condition.", path);
+ }
+ }
+
+ private void cleanUpCompactors() {
+ final String compactorQueuesPath = this.ctx.getZooKeeperRoot() + Constants.ZCOMPACTORS;
+
+ var zoorw = this.ctx.getZooReaderWriter();
+
+ try {
+ var groups = zoorw.getChildren(compactorQueuesPath);
+
+ for (String group : groups) {
+ String qpath = compactorQueuesPath + "/" + group;
+
+ var compactors = zoorw.getChildren(qpath);
+
+ if (compactors.isEmpty()) {
+ deleteEmpty(zoorw, qpath);
+ }
+
+ for (String compactor : compactors) {
+ String cpath = compactorQueuesPath + "/" + group + "/" + compactor;
+ var lockNodes = zoorw.getChildren(compactorQueuesPath + "/" + group + "/" + compactor);
+ if (lockNodes.isEmpty()) {
+ deleteEmpty(zoorw, cpath);
+ }
+ }
+ }
+
+ } catch (KeeperException | RuntimeException e) {
+ LOG.warn("Failed to clean up compactors", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+
+}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
index 60b25d42dd,86bc06930c..136f45c73b
--- a/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
@@@ -158,70 -156,72 +158,70 @@@ public class RecoveryManager
}
}
- public boolean recoverLogs(KeyExtent extent, Collection<Collection<String>> walogs)
- throws IOException {
+ public boolean recoverLogs(KeyExtent extent, Collection<LogEntry> walogs) throws IOException {
boolean recoveryNeeded = false;
- for (Collection<String> logs : walogs) {
- for (String walog : logs) {
-
- Path switchedWalog = VolumeUtil.switchVolume(new Path(walog), FileType.WAL,
- manager.getContext().getVolumeReplacements());
- if (switchedWalog != null) {
- // replaces the volume used for sorting, but do not change entry in metadata table. When
- // the tablet loads it will change the metadata table entry. If
- // the tablet has the same replacement config, then it will find the sorted log.
- log.info("Volume replaced {} -> {}", walog, switchedWalog);
- walog = switchedWalog.toString();
- }
+ for (LogEntry entry : walogs) {
+ String walog = entry.filename;
+
- Path switchedWalog = VolumeUtil.switchVolume(walog, FileType.WAL,
++ Path switchedWalog = VolumeUtil.switchVolume(new Path(walog), FileType.WAL,
+ manager.getContext().getVolumeReplacements());
+ if (switchedWalog != null) {
+ // replaces the volume used for sorting, but do not change entry in metadata table. When
+ // the tablet loads it will change the metadata table entry. If
+ // the tablet has the same replacement config, then it will find the sorted log.
+ log.info("Volume replaced {} -> {}", walog, switchedWalog);
+ walog = switchedWalog.toString();
+ }
- String[] parts = walog.split("/");
- String sortId = parts[parts.length - 1];
- String filename = new Path(walog).toString();
- String dest = RecoveryPath.getRecoveryPath(new Path(filename)).toString();
+ String[] parts = walog.split("/");
+ String sortId = parts[parts.length - 1];
+ String filename = new Path(walog).toString();
+ String dest = RecoveryPath.getRecoveryPath(new Path(filename)).toString();
+
+ boolean sortQueued;
+ synchronized (this) {
+ sortQueued = sortsQueued.contains(sortId);
+ }
- boolean sortQueued;
+ if (sortQueued
+ && zooCache.get(manager.getZooKeeperRoot() + Constants.ZRECOVERY + "/" + sortId)
+ == null) {
synchronized (this) {
- sortQueued = sortsQueued.contains(sortId);
+ sortsQueued.remove(sortId);
}
+ }
- if (sortQueued
- && zooCache.get(manager.getZooKeeperRoot() + Constants.ZRECOVERY + "/" + sortId)
- == null) {
- synchronized (this) {
- sortsQueued.remove(sortId);
- }
+ if (exists(SortedLogState.getFinishedMarkerPath(dest))) {
+ synchronized (this) {
+ closeTasksQueued.remove(sortId);
+ recoveryDelay.remove(sortId);
+ sortsQueued.remove(sortId);
}
+ continue;
+ }
- if (exists(SortedLogState.getFinishedMarkerPath(dest))) {
- synchronized (this) {
- closeTasksQueued.remove(sortId);
- recoveryDelay.remove(sortId);
- sortsQueued.remove(sortId);
+ recoveryNeeded = true;
+ synchronized (this) {
+ if (!closeTasksQueued.contains(sortId) && !sortsQueued.contains(sortId)) {
+ AccumuloConfiguration aconf = manager.getConfiguration();
+ LogCloser closer = Property.createInstanceFromPropertyName(aconf,
+ Property.MANAGER_WAL_CLOSER_IMPLEMENTATION, LogCloser.class, new HadoopLogCloser());
+ Long delay = recoveryDelay.get(sortId);
+ if (delay == null) {
+ delay = aconf.getTimeInMillis(Property.MANAGER_RECOVERY_DELAY);
+ } else {
+ delay = Math.min(2 * delay, 1000 * 60 * 5L);
}
- continue;
- }
- recoveryNeeded = true;
- synchronized (this) {
- if (!closeTasksQueued.contains(sortId) && !sortsQueued.contains(sortId)) {
- AccumuloConfiguration aconf = manager.getConfiguration();
- LogCloser closer = Property.createInstanceFromPropertyName(aconf,
- Property.MANAGER_WAL_CLOSER_IMPLEMENTATION, LogCloser.class, new HadoopLogCloser());
- Long delay = recoveryDelay.get(sortId);
- if (delay == null) {
- delay = aconf.getTimeInMillis(Property.MANAGER_RECOVERY_DELAY);
- } else {
- delay = Math.min(2 * delay, 1000 * 60 * 5L);
- }
-
- log.info("Starting recovery of {} (in : {}s), tablet {} holds a reference", filename,
- (delay / 1000), extent);
-
- ScheduledFuture<?> future = executor.schedule(
- new LogSortTask(closer, filename, dest, sortId), delay, TimeUnit.MILLISECONDS);
- ThreadPools.watchNonCriticalScheduledTask(future);
- closeTasksQueued.add(sortId);
- recoveryDelay.put(sortId, delay);
- }
+ log.info("Starting recovery of {} (in : {}s), tablet {} holds a reference", filename,
+ (delay / 1000), extent);
+
+ ScheduledFuture<?> future = executor.schedule(
+ new LogSortTask(closer, filename, dest, sortId), delay, TimeUnit.MILLISECONDS);
+ ThreadPools.watchNonCriticalScheduledTask(future);
+ closeTasksQueued.add(sortId);
+ recoveryDelay.put(sortId, delay);
}
}
}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/state/MergeStats.java
index eee0dd0ba6,27409a1a9f..59e4b087f1
--- a/server/manager/src/main/java/org/apache/accumulo/manager/state/MergeStats.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/state/MergeStats.java
@@@ -256,14 -189,12 +191,13 @@@ public class MergeStats
return false;
}
- prevExtent = tls.extent;
+ prevExtent = tm.getExtent();
- verify.update(tls.extent, tls.getState(manager.onlineTabletServers()));
+ Set<TServerInstance> liveTServers = manager.onlineTabletServers();
- verify.update(tm.getExtent(), TabletState.compute(tm, liveTServers), tm.hasChopped(),
- !tm.getLogs().isEmpty());
++ verify.update(tm.getExtent(), TabletState.compute(tm, liveTServers));
// stop when we've seen the tablet just beyond our range
- if (tls.extent.prevEndRow() != null && extent.endRow() != null
- && tls.extent.prevEndRow().compareTo(extent.endRow()) > 0) {
+ if (tm.getExtent().prevEndRow() != null && extent.endRow() != null
+ && tm.getExtent().prevEndRow().compareTo(extent.endRow()) > 0) {
break;
}
}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/CleanUpBulkImport.java
index 139762d557,e613d53f93..7ccfbcdf95
--- a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/CleanUpBulkImport.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/CleanUpBulkImport.java
@@@ -66,15 -59,17 +66,15 @@@ public class CleanUpBulkImport extends
ample.removeBulkLoadInProgressFlag(
"/" + bulkDir.getParent().getName() + "/" + bulkDir.getName());
ample.putGcFileAndDirCandidates(info.tableId,
- Collections.singleton(new ReferenceFile(info.tableId, bulkDir.toString())));
+ Collections.singleton(new ReferenceFile(info.tableId, bulkDir)));
- if (info.tableState == TableState.ONLINE) {
- Text firstSplit = info.firstSplit == null ? null : new Text(info.firstSplit);
- Text lastSplit = info.lastSplit == null ? null : new Text(info.lastSplit);
+ Text firstSplit = info.firstSplit == null ? null : new Text(info.firstSplit);
+ Text lastSplit = info.lastSplit == null ? null : new Text(info.lastSplit);
- log.debug("{} removing the metadata table markers for loaded files in range {} {}",
- FateTxId.formatTid(tid), firstSplit, lastSplit);
+ log.debug("{} removing the metadata table markers for loaded files in range {} {}",
+ FateTxId.formatTid(tid), firstSplit, lastSplit);
+ removeBulkLoadEntries(ample, info.tableId, tid, firstSplit, lastSplit);
- ample.removeBulkLoadEntries(info.tableId, tid, firstSplit, lastSplit);
- }
Utils.unreserveHdfsDirectory(manager, info.sourceDir, tid);
Utils.getReadLock(manager, info.tableId, tid).unlock();
// delete json renames and mapping files
diff --cc server/manager/src/test/java/org/apache/accumulo/manager/split/SplitterTest.java
index 42767f0758,0000000000..75c0f16bea
mode 100644,000000..100644
--- a/server/manager/src/test/java/org/apache/accumulo/manager/split/SplitterTest.java
+++ b/server/manager/src/test/java/org/apache/accumulo/manager/split/SplitterTest.java
@@@ -1,116 -1,0 +1,123 @@@
+/*
+ * 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.split;
+
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.createNiceMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
++import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.StoredTabletFile;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.util.cache.Caches;
+import org.apache.accumulo.core.util.threads.ThreadPools;
+import org.apache.accumulo.server.ServerContext;
++import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.Test;
+
+public class SplitterTest {
+
+ @Test
+ public void testIsSplittable() {
+ ThreadPools threadPools = createNiceMock(ThreadPools.class);
+ replay(threadPools);
+ ServerContext context = createNiceMock(ServerContext.class);
+ expect(context.threadPools()).andReturn(threadPools).anyTimes();
+ expect(context.getCaches()).andReturn(Caches.getInstance()).anyTimes();
+ replay(context);
+
+ var splitter = new Splitter(context);
+
+ KeyExtent ke1 = new KeyExtent(TableId.of("1"), new Text("m"), null);
+ KeyExtent ke2 = new KeyExtent(TableId.of("1"), null, new Text("m"));
+
+ Set<StoredTabletFile> files1 = new HashSet<>();
- files1.add(new StoredTabletFile(
- "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000070.rf"));
- files1.add(new StoredTabletFile(
- "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000072.rf"));
++ files1.add(new ReferencedTabletFile(
++ new Path("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000070.rf")).insert());
++ files1.add(new ReferencedTabletFile(
++ new Path("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000072.rf")).insert());
+
+ TabletMetadata tabletMeta1 = createMock(TabletMetadata.class);
+ expect(tabletMeta1.getExtent()).andReturn(ke1).anyTimes();
+ expect(tabletMeta1.getFiles()).andReturn(files1).times(3);
+ replay(tabletMeta1);
+
+ TabletMetadata tabletMeta2 = createMock(TabletMetadata.class);
+ expect(tabletMeta2.getExtent()).andReturn(ke2).anyTimes();
+ replay(tabletMeta2);
+
+ assertTrue(splitter.isSplittable(tabletMeta1));
+ assertTrue(splitter.isSplittable(tabletMeta2));
+
+ splitter.addSplitStarting(ke1);
+
+ assertFalse(splitter.isSplittable(tabletMeta1));
+ assertTrue(splitter.isSplittable(tabletMeta2));
+
+ splitter.removeSplitStarting(ke1);
+
+ assertTrue(splitter.isSplittable(tabletMeta1));
+ assertTrue(splitter.isSplittable(tabletMeta2));
+
+ splitter.rememberUnsplittable(tabletMeta1);
+
+ assertFalse(splitter.isSplittable(tabletMeta1));
+ assertTrue(splitter.isSplittable(tabletMeta2));
+
+ // tabletMeta1 is currently unsplittable. Adding a file
+ // to it's file set should cause it to be removed from
+ // the unsplittable set of tablets, becoming splittable
+ // again.
- files1.add(new StoredTabletFile(
- "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000071.rf"));
++ files1.add(new ReferencedTabletFile(
++ new Path("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000071.rf")).insert());
+ assertTrue(splitter.isSplittable(tabletMeta1));
+
+ // when a tablets files change it should become a candidate for inspection
+ Set<StoredTabletFile> files2 = Set.of(
- new StoredTabletFile("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000070.rf"),
- new StoredTabletFile("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000072.rf"),
- new StoredTabletFile(
- "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000073.rf"));
++ new ReferencedTabletFile(
++ new Path("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000070.rf"))
++ .insert(),
++ new ReferencedTabletFile(
++ new Path("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000072.rf"))
++ .insert(),
++ new ReferencedTabletFile(
++ new Path("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000073.rf"))
++ .insert());
+ TabletMetadata tabletMeta3 = createMock(TabletMetadata.class);
+ expect(tabletMeta3.getExtent()).andReturn(ke1).anyTimes();
+ expect(tabletMeta3.getFiles()).andReturn(files2).anyTimes();
+ replay(tabletMeta3);
+
+ assertTrue(splitter.isSplittable(tabletMeta3));
+ assertTrue(splitter.isSplittable(tabletMeta2));
+
+ verify(threadPools, context, tabletMeta1, tabletMeta2, tabletMeta3);
+ }
+
+}
diff --cc server/manager/src/test/java/org/apache/accumulo/manager/tableOps/split/UpdateTabletsTest.java
index 08f2151d5f,0000000000..68f1a85424
mode 100644,000000..100644
--- a/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/split/UpdateTabletsTest.java
+++ b/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/split/UpdateTabletsTest.java
@@@ -1,116 -1,0 +1,119 @@@
+/*
+ * 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.split;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
++import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.StoredTabletFile;
+import org.apache.accumulo.core.metadata.schema.DataFileValue;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.server.util.FileUtil;
++import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.easymock.EasyMock;
+import org.junit.jupiter.api.Test;
+
+public class UpdateTabletsTest {
+
+ StoredTabletFile newSTF(int fileNum) {
- return new StoredTabletFile(
- "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F00000" + fileNum + ".rf");
++ return new ReferencedTabletFile(new Path(
++ "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F00000" + fileNum + ".rf"))
++ .insert();
+ }
+
+ FileUtil.FileInfo newFileInfo(String start, String end) {
+ return new FileUtil.FileInfo(new Key(start), new Key(end));
+ }
+
+ // When a tablet splits its files are partitioned among the new children tablets. This test
+ // exercises the partitioning code.
+ @Test
+ public void testFileParitioning() {
+
+ var file1 = newSTF(1);
+ var file2 = newSTF(2);
+ var file3 = newSTF(3);
+ var file4 = newSTF(4);
+
+ var tabletFiles =
+ Map.of(file1, new DataFileValue(1000, 100, 20), file2, new DataFileValue(2000, 200, 50),
+ file3, new DataFileValue(4000, 400), file4, new DataFileValue(4000, 400));
+
+ var ke1 = new KeyExtent(TableId.of("1"), new Text("m"), null);
+ var ke2 = new KeyExtent(TableId.of("1"), new Text("r"), new Text("m"));
+ var ke3 = new KeyExtent(TableId.of("1"), new Text("v"), new Text("r"));
+ var ke4 = new KeyExtent(TableId.of("1"), null, new Text("v"));
+
+ var firstAndLastKeys = Map.of(file2, newFileInfo("m", "r"), file3, newFileInfo("g", "x"), file4,
+ newFileInfo("s", "v"));
+
+ var ke1Expected = Map.of(file1, new DataFileValue(250, 25, 20), file2,
+ new DataFileValue(1000, 100, 50), file3, new DataFileValue(1000, 100));
+ var ke2Expected = Map.of(file1, new DataFileValue(250, 25, 20), file2,
+ new DataFileValue(1000, 100, 50), file3, new DataFileValue(1000, 100));
+ var ke3Expected = Map.of(file1, new DataFileValue(250, 25, 20), file3,
+ new DataFileValue(1000, 100), file4, new DataFileValue(4000, 400));
+ var ke4Expected =
+ Map.of(file1, new DataFileValue(250, 25, 20), file3, new DataFileValue(1000, 100));
+
+ var expected = Map.of(ke1, ke1Expected, ke2, ke2Expected, ke3, ke3Expected, ke4, ke4Expected);
+
+ Set<KeyExtent> newExtents = Set.of(ke1, ke2, ke3, ke4);
+
+ TabletMetadata tabletMeta = EasyMock.createMock(TabletMetadata.class);
+ EasyMock.expect(tabletMeta.getFilesMap()).andReturn(tabletFiles).anyTimes();
+ EasyMock.replay(tabletMeta);
+
+ Map<KeyExtent,Map<StoredTabletFile,DataFileValue>> results =
+ UpdateTablets.getNewTabletFiles(newExtents, tabletMeta, firstAndLastKeys::get);
+
+ assertEquals(expected.keySet(), results.keySet());
+ expected.forEach(((extent, files) -> {
+ assertEquals(files, results.get(extent));
+ }));
+
+ // Test a tablet with no files going to it
+
+ var tabletFiles2 = Map.of(file2, tabletFiles.get(file2), file4, tabletFiles.get(file4));
+ ke1Expected = Map.of(file2, new DataFileValue(1000, 100, 50));
+ ke2Expected = Map.of(file2, new DataFileValue(1000, 100, 50));
+ ke3Expected = Map.of(file4, new DataFileValue(4000, 400));
+ ke4Expected = Map.of();
+ expected = Map.of(ke1, ke1Expected, ke2, ke2Expected, ke3, ke3Expected, ke4, ke4Expected);
+
+ tabletMeta = EasyMock.createMock(TabletMetadata.class);
+ EasyMock.expect(tabletMeta.getFilesMap()).andReturn(tabletFiles2).anyTimes();
+ EasyMock.replay(tabletMeta);
+
+ Map<KeyExtent,Map<StoredTabletFile,DataFileValue>> results2 =
+ UpdateTablets.getNewTabletFiles(newExtents, tabletMeta, firstAndLastKeys::get);
+ assertEquals(expected.keySet(), results2.keySet());
+ expected.forEach(((extent, files) -> {
+ assertEquals(files, results2.get(extent));
+ }));
+
+ }
+}
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
index 60ffaaeb4b,f3337a4e25..789649758c
--- 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
@@@ -1352,155 -2179,28 +1351,154 @@@ public class Tablet extends TabletBase
return timer.getTabletStats();
}
- private static String createTabletDirectoryName(ServerContext context, Text endRow) {
- if (endRow == null) {
- return ServerColumnFamily.DEFAULT_TABLET_DIR_NAME;
- } else {
- UniqueNameAllocator namer = context.getUniqueNameAllocator();
- return Constants.GENERATED_TABLET_DIRECTORY_PREFIX + namer.getNextName();
- }
+ public boolean isOnDemand() {
+ // TODO a change in the hosting goal could refresh online tablets
+ return getMetadata().getHostingGoal() == TabletHostingGoal.ONDEMAND;
}
- public Set<Long> getBulkIngestedTxIds() {
- return bulkImported.keySet();
- }
+ // The purpose of this lock is to prevent race conditions between concurrent refresh RPC calls and
+ // between minor compactions and refresh calls.
+ private final Lock refreshLock = new ReentrantLock();
+
+ void bringMinorCompactionOnline(ReferencedTabletFile tmpDatafile,
+ ReferencedTabletFile newDatafile, DataFileValue dfv, CommitSession commitSession,
+ long flushId) {
+ Optional<StoredTabletFile> newFile;
+ // rename before putting in metadata table, so files in metadata table should
+ // always exist
+ boolean attemptedRename = false;
+ VolumeManager vm = getTabletServer().getContext().getVolumeManager();
+ do {
+ try {
+ if (dfv.getNumEntries() == 0) {
+ log.debug("No data entries so delete temporary file {}", tmpDatafile);
+ vm.deleteRecursively(tmpDatafile.getPath());
+ } else {
+ if (!attemptedRename && vm.exists(newDatafile.getPath())) {
+ log.warn("Target data file already exist {}", newDatafile);
+ throw new RuntimeException("File unexpectedly exists " + newDatafile.getPath());
+ }
+ // the following checks for spurious rename failures that succeeded but gave an IoE
+ if (attemptedRename && vm.exists(newDatafile.getPath())
+ && !vm.exists(tmpDatafile.getPath())) {
+ // seems like previous rename succeeded, so break
+ break;
+ }
+ attemptedRename = true;
+ ScanfileManager.rename(vm, tmpDatafile.getPath(), newDatafile.getPath());
+ }
+ break;
+ } catch (IOException ioe) {
+ log.warn("Tablet " + getExtent() + " failed to rename " + newDatafile
+ + " after MinC, will retry in 60 secs...", ioe);
+ sleepUninterruptibly(1, TimeUnit.MINUTES);
+ }
+ } while (true);
+
+ // The refresh lock must be held for the metadata write that adds the new file to the tablet.
+ // This prevents a concurrent refresh operation from pulling in the new tablet file before the
+ // in memory map reference related to the file is deactivated. Scans should use one of the in
+ // memory map or the new file, never both.
+ Preconditions.checkState(!getLogLock().isHeldByCurrentThread());
+ refreshLock.lock();
+ try {
+ // Can not hold tablet lock while acquiring the log lock. The following check is there to
+ // prevent deadlock.
- Preconditions.checkState(!Thread.holdsLock(this));
+ getLogLock().lock();
+ // do not place any code here between lock and try
+ try {
+ // The following call pairs with tablet.finishClearingUnusedLogs() later in this block. If
+ // moving where the following method is called, examine it and finishClearingUnusedLogs()
+ // before moving.
+ Set<String> unusedWalLogs = beginClearingUnusedLogs();
+ // the order of writing to metadata and walog is important in the face of machine/process
+ // failures need to write to metadata before writing to walog, when things are done in the
+ // reverse order data could be lost... the minor compaction start event should be written
+ // before the following metadata write is made
+
+ newFile = updateTabletDataFile(commitSession.getMaxCommittedTime(), newDatafile, dfv,
+ unusedWalLogs, flushId);
+
+ finishClearingUnusedLogs();
+ } finally {
+ getLogLock().unlock();
+ }
+
+ // Without the refresh lock, if a refresh happened here it could make the new file written to
+ // the metadata table above available for scans while the in memory map from which the file
+ // was produced is still available for scans
- public void cleanupBulkLoadedFiles(Set<Long> tids) {
- bulkImported.keySet().removeAll(tids);
+ do {
+ try {
+ // the purpose of making this update use the new commit session, instead of the old one
+ // passed in, is because the new one will reference the logs used by current memory...
+ getTabletServer().minorCompactionFinished(getTabletMemory().getCommitSession(),
+ commitSession.getWALogSeq() + 2);
+ break;
+ } catch (IOException e) {
+ log.error("Failed to write to write-ahead log " + e.getMessage() + " will retry", e);
+ sleepUninterruptibly(1, TimeUnit.SECONDS);
+ }
+ } while (true);
+
+ refreshMetadata(RefreshPurpose.MINC_COMPLETION);
+ } finally {
+ refreshLock.unlock();
+ }
+ TabletLogger.flushed(getExtent(), newFile);
+
+ long splitSize = getTableConfiguration().getAsBytes(Property.TABLE_SPLIT_THRESHOLD);
+ if (dfv.getSize() > splitSize) {
+ log.debug(String.format("Minor Compaction wrote out file larger than split threshold."
+ + " split threshold = %,d file size = %,d", splitSize, dfv.getSize()));
+ }
}
- public String getDirName() {
- return dirName;
+ public enum RefreshPurpose {
+ MINC_COMPLETION, REFRESH_RPC, FLUSH_ID_UPDATE
}
- public Compactable asCompactable() {
- return compactable;
+ public void refreshMetadata(RefreshPurpose refreshPurpose) {
+ refreshLock.lock();
+ try {
+
+ // do not want to hold tablet lock while doing metadata read as this could negatively impact
+ // scans
+ TabletMetadata tabletMetadata = getContext().getAmple().readTablet(getExtent());
+
+ synchronized (this) {
+ latestMetadata = tabletMetadata;
+
+ if (refreshPurpose == RefreshPurpose.MINC_COMPLETION) {
+ // Atomically replace the in memory map with the new file. Before this synch block a scan
+ // starting would see the in memory map. After this synch block it should see the file in
+ // the tabletMetadata. Scans sync on the tablet also, so they can not be in this code
+ // block at the same time.
+
+ lastLocation = null;
+ tabletMemory.finishedMinC();
+
+ // the files and in memory map changed, incrementing this will cause scans to switch data
+ // sources
+ dataSourceDeletions.incrementAndGet();
+
+ // important to call this after updating latestMetadata and tabletMemory
+ computeNumEntries();
+ } else if (!latestMetadata.getFilesMap().equals(tabletMetadata.getFilesMap())) {
+
+ // the files changed, incrementing this will cause scans to switch data sources
+ dataSourceDeletions.incrementAndGet();
+
+ // important to call this after updating latestMetadata
+ computeNumEntries();
+ }
+ }
+ } finally {
+ refreshLock.unlock();
+ }
+
+ if (refreshPurpose == RefreshPurpose.REFRESH_RPC) {
+ scanfileManager.removeFilesAfterScan(getMetadata().getScans());
+ }
}
}
diff --cc server/tserver/src/test/java/org/apache/accumulo/tserver/compaction/CompactableUtilsTest.java
index b288af0f99,84229c7a21..f8dfd4391b
--- a/server/tserver/src/test/java/org/apache/accumulo/tserver/compaction/CompactableUtilsTest.java
+++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/compaction/CompactableUtilsTest.java
@@@ -32,8 -32,8 +32,8 @@@ public class CompactableUtilsTest
ReferencedTabletFile expected = new ReferencedTabletFile(
new Path("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000070.rf"));
ReferencedTabletFile tmpFile =
- new ReferencedTabletFile(new Path(expected.getMetaInsert() + "_tmp"));
+ new ReferencedTabletFile(new Path(expected.getNormalizedPathStr() + "_tmp"));
- ReferencedTabletFile dest = CompactableUtils.computeCompactionFileDest(tmpFile);
+ ReferencedTabletFile dest = TabletNameGenerator.computeCompactionFileDest(tmpFile);
assertEquals(expected, dest);
}
diff --cc shell/src/test/java/org/apache/accumulo/shell/commands/ListTabletsCommandTest.java
index e080d29f13,583819b96b..c9f3f58836
--- a/shell/src/test/java/org/apache/accumulo/shell/commands/ListTabletsCommandTest.java
+++ b/shell/src/test/java/org/apache/accumulo/shell/commands/ListTabletsCommandTest.java
@@@ -33,22 -31,16 +33,24 @@@ import java.util.stream.Stream
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.admin.InstanceOperations;
import org.apache.accumulo.core.client.admin.TableOperations;
+import org.apache.accumulo.core.client.admin.TabletHostingGoal;
import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.TabletInformationImpl;
+import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.core.metadata.TabletState;
-import org.apache.accumulo.core.metadata.schema.TabletMetadata.Location;
++import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.StoredTabletFile;
+import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.core.metadata.schema.DataFileValue;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.shell.Shell;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Options;
++import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.easymock.EasyMock;
import org.junit.jupiter.api.Test;
@@@ -125,46 -113,6 +127,50 @@@ public class ListTabletsCommandTest
public void mockTest() throws Exception {
ListTabletsCommand cmd = new TestListTabletsCommand();
+ TableId tableId = TableId.of("123");
+
+ TServerInstance ser1 = new TServerInstance(HostAndPort.fromParts("server1", 8555), "s001");
+ TServerInstance ser2 = new TServerInstance(HostAndPort.fromParts("server2", 2354), "s002");
+
- StoredTabletFile sf11 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-dir1/sf11.rf");
++ StoredTabletFile sf11 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-dir1/sf11.rf")).insert();
+ DataFileValue dfv11 = new DataFileValue(5643, 89);
+
- StoredTabletFile sf12 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-dir1/sf12.rf");
++ StoredTabletFile sf12 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-dir1/sf12.rf")).insert();
+ DataFileValue dfv12 = new DataFileValue(379963, 1027);
+
- StoredTabletFile sf21 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-dir2/sf21.rf");
++ StoredTabletFile sf21 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-dir2/sf21.rf")).insert();
+ DataFileValue dfv21 = new DataFileValue(5323, 142);
+
- StoredTabletFile sf31 = new StoredTabletFile("hdfs://nn1/acc/tables/1/t-dir3/sf31.rf");
++ StoredTabletFile sf31 =
++ new ReferencedTabletFile(new Path("hdfs://nn1/acc/tables/1/t-dir3/sf31.rf")).insert();
+ DataFileValue dfv31 = new DataFileValue(95832L, 231);
+
+ KeyExtent extent = new KeyExtent(tableId, new Text("d"), null);
+
+ LogEntry le1 = new LogEntry(extent, 55, "lf1");
+ LogEntry le2 = new LogEntry(extent, 57, "lf2");
+
+ TabletMetadata tm1 = TabletMetadata.builder(extent).putHostingGoal(TabletHostingGoal.ONDEMAND)
+ .putLocation(TabletMetadata.Location.current(ser1)).putFile(sf11, dfv11)
+ .putFile(sf12, dfv12).putWal(le1).putDirName("t-dir1").build();
+
+ extent = new KeyExtent(tableId, new Text("k"), new Text("e"));
+ TabletMetadata tm2 = TabletMetadata.builder(extent).putHostingGoal(TabletHostingGoal.ALWAYS)
+ .putLocation(TabletMetadata.Location.current(ser2)).putFile(sf21, dfv21)
+ .putDirName("t-dir2").build(LOGS);
+
+ extent = new KeyExtent(tableId, null, new Text("l"));
+ TabletMetadata tm3 = TabletMetadata.builder(extent).putHostingGoal(TabletHostingGoal.NEVER)
+ .putFile(sf31, dfv31).putWal(le1).putWal(le2).putDirName("t-dir3").build(LOCATION);
+
+ TabletInformationImpl[] tabletInformation = new TabletInformationImpl[3];
+ tabletInformation[0] = new TabletInformationImpl(tm1, "HOSTED");
+ tabletInformation[1] = new TabletInformationImpl(tm2, "HOSTED");
+ tabletInformation[2] = new TabletInformationImpl(tm3, "UNASSIGNED");
+
AccumuloClient client = EasyMock.createMock(AccumuloClient.class);
ClientContext context = EasyMock.createMock(ClientContext.class);
TableOperations tableOps = EasyMock.createMock(TableOperations.class);
diff --cc test/src/main/java/org/apache/accumulo/test/functional/MergeIT.java
index 0bfa4b545c,c8e8f93de5..94e5927927
--- a/test/src/main/java/org/apache/accumulo/test/functional/MergeIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/MergeIT.java
@@@ -32,29 -37,37 +37,44 @@@ import java.util.TreeSet
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
+ import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Scanner;
+ import org.apache.accumulo.core.client.admin.CompactionConfig;
import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+import org.apache.accumulo.core.client.admin.TabletHostingGoal;
import org.apache.accumulo.core.client.admin.TimeType;
+import org.apache.accumulo.core.clientImpl.TabletHostingGoalUtil;
+ 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.Range;
+ import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.metadata.MetadataTable;
+ import org.apache.accumulo.core.metadata.StoredTabletFile;
+ import org.apache.accumulo.core.metadata.schema.DataFileValue;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.HostingColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
import org.apache.accumulo.core.security.Authorizations;
+ import org.apache.accumulo.core.util.FastFormat;
import org.apache.accumulo.core.util.Merge;
import org.apache.accumulo.harness.AccumuloClusterHarness;
+ import org.apache.accumulo.test.TestIngest;
+ import org.apache.accumulo.test.TestIngest.IngestParams;
+ import org.apache.accumulo.test.VerifyIngest;
+ import org.apache.accumulo.test.VerifyIngest.VerifyParams;
import org.apache.hadoop.io.Text;
import org.junit.jupiter.api.Test;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+import com.google.common.collect.Iterables;
+
public class MergeIT extends AccumuloClusterHarness {
+ private static final Logger log = LoggerFactory.getLogger(MergeIT.class);
+
SortedSet<Text> splits(String[] points) {
SortedSet<Text> result = new TreeSet<>();
for (String point : points) {
diff --cc test/src/main/java/org/apache/accumulo/test/functional/SplitRecoveryIT.java
index 76df5852e3,05da1eec15..c783e8156e
--- a/test/src/main/java/org/apache/accumulo/test/functional/SplitRecoveryIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/SplitRecoveryIT.java
@@@ -37,6 -37,6 +37,7 @@@ import java.util.UUID
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Scanner;
++import org.apache.accumulo.core.client.admin.TabletHostingGoal;
import org.apache.accumulo.core.client.admin.TimeType;
import org.apache.accumulo.core.clientImpl.ScannerImpl;
import org.apache.accumulo.core.conf.SiteConfiguration;
@@@ -165,7 -163,7 +166,8 @@@ public class SplitRecoveryIT extends Co
String dirName = "dir_" + i;
String tdir =
context.getTablesDirs().iterator().next() + "/" + extent.tableId() + "/" + dirName;
-- MetadataTableUtil.addTablet(extent, dirName, context, TimeType.LOGICAL, zl);
++ MetadataTableUtil.addTablet(extent, dirName, context, TimeType.LOGICAL, zl,
++ TabletHostingGoal.ONDEMAND);
SortedMap<ReferencedTabletFile,DataFileValue> dataFiles = new TreeMap<>();
dataFiles.put(new ReferencedTabletFile(new Path(tdir + "/" + RFile.EXTENSION + "_000_000")),
new DataFileValue(1000017 + i, 10000 + i));
diff --cc test/src/main/java/org/apache/accumulo/test/manager/MergeStateIT.java
index ac618b01e0,003b387f1b..6b199eb860
--- a/test/src/main/java/org/apache/accumulo/test/manager/MergeStateIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/manager/MergeStateIT.java
@@@ -45,13 -38,10 +45,12 @@@ import org.apache.accumulo.core.manager
import org.apache.accumulo.core.manager.thrift.ManagerState;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.TServerInstance;
-import org.apache.accumulo.core.metadata.TabletLocationState;
+import org.apache.accumulo.core.metadata.TabletState;
import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
- import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CurrentLocationColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
+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.security.Authorizations;
import org.apache.accumulo.core.security.TablePermission;
@@@ -143,24 -116,17 +142,25 @@@ public class MergeStateIT extends Confi
MetadataTable.NAME, TablePermission.WRITE);
BatchWriter bw = accumuloClient.createBatchWriter(MetadataTable.NAME);
- // Create a fake METADATA table with these splits
- String[] splits = {"a", "e", "j", "o", "t", "z"};
- // create metadata for a table "t" with the splits above
- TableId tableId = TableId.of("t");
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("a"));
+ splits.add(new Text("e"));
+ splits.add(new Text("j"));
+ splits.add(new Text("o"));
+ splits.add(new Text("t"));
+ splits.add(new Text("z"));
+ NewTableConfiguration ntc = new NewTableConfiguration();
+ ntc.withSplits(splits);
+ accumuloClient.tableOperations().create("merge_test_table");
+ TableId tableId =
+ TableId.of(accumuloClient.tableOperations().tableIdMap().get("merge_test_table"));
+
Text pr = null;
- for (String s : splits) {
- Text split = new Text(s);
+ for (Text split : splits) {
Mutation prevRow =
TabletColumnFamily.createPrevRowMutation(new KeyExtent(tableId, split, pr));
- ChoppedColumnFamily.CHOPPED_COLUMN.put(prevRow, new Value("junk"));
+ prevRow.put(CurrentLocationColumnFamily.NAME, new Text("123456"),
+ new Value("127.0.0.1:1234"));
bw.addMutation(prevRow);
pr = split;
}
@@@ -248,11 -206,8 +239,10 @@@
private MergeStats scan(MockCurrentState state, TabletStateStore metaDataStateStore) {
MergeStats stats = new MergeStats(state.mergeInfo);
stats.getMergeInfo().setState(MergeState.WAITING_FOR_OFFLINE);
- for (TabletLocationState tss : metaDataStateStore) {
- stats.update(tss.extent, tss.getState(state.onlineTabletServers()));
+ for (TabletManagement tm : metaDataStateStore) {
+ TabletMetadata tabletMetadata = tm.getTabletMetadata();
+ stats.update(tm.getTabletMetadata().getExtent(),
- TabletState.compute(tabletMetadata, state.onlineTabletServers()),
- tm.getTabletMetadata().hasChopped(), false);
++ TabletState.compute(tabletMetadata, state.onlineTabletServers()));
}
return stats;
}
diff --cc test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java
index f32ddbdd60,bd5897c1ac..88eb567693
--- a/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java
+++ b/test/src/main/java/org/apache/accumulo/test/performance/NullTserver.java
@@@ -111,6 -116,15 +111,11 @@@ public class NullTserver
return new UpdateErrors(new HashMap<>(), new ArrayList<>(), new HashMap<>());
}
+ @Override
+ public boolean cancelUpdate(TInfo tinfo, long updateID) throws TException {
+ return true;
+ }
+
- @Override
- public void loadFiles(TInfo tinfo, TCredentials credentials, long tid, String dir,
- Map<TKeyExtent,Map<String,DataFileInfo>> fileMap, boolean setTime) {}
-
@Override
public void closeMultiScan(TInfo tinfo, long scanID) {}