You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by do...@apache.org on 2024/08/16 21:20:55 UTC
(accumulo) branch elasticity updated (00cb09f957 -> f5396dfa85)
This is an automated email from the ASF dual-hosted git repository.
domgarguilo pushed a change to branch elasticity
in repository https://gitbox.apache.org/repos/asf/accumulo.git
from 00cb09f957 Merge remote-tracking branch 'upstream/main' into elasticity
add fd111a22a6 Ensure resources are closed in GarbageCollectWriteAheadLogs (#4790)
add eda39bb1c1 Merge remote-tracking branch 'upstream/2.1' into main
new f5396dfa85 Merge remote-tracking branch 'upstream/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:
.../accumulo/gc/GarbageCollectWriteAheadLogs.java | 28 +++++++-----
.../gc/GarbageCollectWriteAheadLogsTest.java | 51 +++++++---------------
2 files changed, 34 insertions(+), 45 deletions(-)
(accumulo) 01/01: Merge remote-tracking branch 'upstream/main' into elasticity
Posted by do...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
domgarguilo pushed a commit to branch elasticity
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit f5396dfa85b3e127c132a549cd98c53dfaca6b05
Merge: 00cb09f957 eda39bb1c1
Author: Dom Garguilo <do...@apache.org>
AuthorDate: Fri Aug 16 17:19:54 2024 -0400
Merge remote-tracking branch 'upstream/main' into elasticity
.../accumulo/gc/GarbageCollectWriteAheadLogs.java | 28 +++++++-----
.../gc/GarbageCollectWriteAheadLogsTest.java | 51 +++++++---------------
2 files changed, 34 insertions(+), 45 deletions(-)
diff --cc server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
index 79cbe9535a,c636c34c4b..4d2f1b4dc6
--- a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
@@@ -72,7 -65,8 +72,7 @@@ public class GarbageCollectWriteAheadLo
private final VolumeManager fs;
private final LiveTServerSet liveServers;
private final WalStateManager walMarker;
- private final AtomicBoolean hasCollected;
+ private final AtomicBoolean hasCollected = new AtomicBoolean(false);
- private final Stream<TabletLocationState> store;
/**
* Creates a new GC WAL object.
@@@ -82,26 -76,44 +82,34 @@@
*/
GarbageCollectWriteAheadLogs(final ServerContext context, final VolumeManager fs,
final LiveTServerSet liveServers) {
- this(context, fs, liveServers, new WalStateManager(context), createStore(context));
++ this(context, fs, liveServers, new WalStateManager(context));
+ }
+
+ /**
+ * Creates a new GC WAL object. Meant for testing -- allows for mocked objects.
+ *
+ *
+ * @param context the collection server's context
+ * @param fs volume manager to use
+ * @param liveServers a started LiveTServerSet instance
+ * @param walMarker a WalStateManager instance
- * @param store a stream of TabletLocationState objects
+ */
+ GarbageCollectWriteAheadLogs(final ServerContext context, final VolumeManager fs,
- final LiveTServerSet liveServers, final WalStateManager walMarker,
- final Stream<TabletLocationState> store) {
++ final LiveTServerSet liveServers, final WalStateManager walMarker) {
this.context = context;
this.fs = fs;
this.liveServers = liveServers;
- this.walMarker = createWalStateManager(context);
- this.hasCollected = new AtomicBoolean(false);
- }
-
- @VisibleForTesting
- WalStateManager createWalStateManager(ServerContext context) {
- return new WalStateManager(context);
+ this.walMarker = walMarker;
- this.store = store;
}
- private static Stream<TabletLocationState> createStore(final ServerContext context) {
- var rootStream = TabletStateStore.getStoreForLevel(DataLevel.ROOT, context).stream();
- var metadataStream = TabletStateStore.getStoreForLevel(DataLevel.METADATA, context).stream();
- var userStream = TabletStateStore.getStoreForLevel(DataLevel.USER, context).stream();
- return Streams.concat(rootStream, metadataStream, userStream).onClose(() -> {
- try {
- rootStream.close();
- } finally {
- try {
- metadataStream.close();
- } finally {
- userStream.close();
- }
- }
- });
+ @VisibleForTesting
+ Stream<TabletMetadata> createStore(Set<TServerInstance> liveTServers) {
+ GcWalsFilter walsFilter = new GcWalsFilter(liveTServers);
+
+ return Stream.of(DataLevel.ROOT, DataLevel.METADATA, DataLevel.USER)
+ .map(dataLevel -> context.getAmple().readTablets().forLevel(dataLevel).filter(walsFilter)
+ .fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build())
+ .flatMap(TabletsMetadata::stream);
}
public void collect(GCStatus status) {
diff --cc server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
index 7f4ec5173c,13c264a812..f868fd26d1
--- a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
+++ b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
@@@ -100,26 -94,14 +100,21 @@@ public class GarbageCollectWriteAheadLo
marker.removeWalMarker(server1, id);
EasyMock.expectLastCall().once();
EasyMock.replay(context, fs, marker, tserverSet);
- GarbageCollectWriteAheadLogs gc = new GarbageCollectWriteAheadLogs(context, fs, tserverSet) {
- var gc =
- new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker, tabletOnServer1List) {
- @Override
- protected Map<UUID,Path> getSortedWALogs() {
- return Collections.emptyMap();
- }
- };
++ var gc = new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker) {
+ @Override
+ protected Map<UUID,Path> getSortedWALogs() {
+ return Collections.emptyMap();
+ }
+
- @Override
- WalStateManager createWalStateManager(ServerContext serverContext) {
- return marker;
- }
-
+ @Override
+ Stream<TabletMetadata> createStore(Set<TServerInstance> liveTservers) {
+ return tabletOnServer1List;
+ }
+ };
gc.collect(status);
+ assertThrows(IllegalStateException.class, () -> gc.collect(status),
+ "Should only be able to call collect once");
+
EasyMock.verify(context, fs, marker, tserverSet);
}
@@@ -139,24 -121,14 +134,19 @@@
EasyMock.expect(marker.getAllMarkers()).andReturn(markers).once();
EasyMock.expect(marker.state(server1, id)).andReturn(new Pair<>(WalState.CLOSED, path));
EasyMock.replay(context, marker, tserverSet, fs);
- GarbageCollectWriteAheadLogs gc = new GarbageCollectWriteAheadLogs(context, fs, tserverSet) {
- var gc =
- new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker, tabletOnServer1List) {
- @Override
- protected Map<UUID,Path> getSortedWALogs() {
- return Collections.emptyMap();
- }
- };
++ var gc = new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker) {
+ @Override
+ protected Map<UUID,Path> getSortedWALogs() {
+ return Collections.emptyMap();
+ }
+
- @Override
- WalStateManager createWalStateManager(ServerContext serverContext) {
- return marker;
- }
-
+ @Override
+ Stream<TabletMetadata> createStore(Set<TServerInstance> liveTservers) {
+ return tabletOnServer1List;
+ }
+ };
gc.collect(status);
+
EasyMock.verify(context, marker, tserverSet, fs);
}
@@@ -183,24 -155,14 +173,19 @@@
marker.forget(server2);
EasyMock.expectLastCall().once();
EasyMock.replay(context, fs, marker, tserverSet);
- GarbageCollectWriteAheadLogs gc = new GarbageCollectWriteAheadLogs(context, fs, tserverSet) {
- var gc =
- new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker, tabletOnServer1List) {
- @Override
- protected Map<UUID,Path> getSortedWALogs() {
- return Collections.emptyMap();
- }
- };
++ var gc = new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker) {
+ @Override
+ protected Map<UUID,Path> getSortedWALogs() {
+ return Collections.emptyMap();
+ }
+
- @Override
- WalStateManager createWalStateManager(ServerContext serverContext) {
- return marker;
- }
-
+ @Override
+ Stream<TabletMetadata> createStore(Set<TServerInstance> liveTservers) {
+ return tabletOnServer1List;
+ }
+ };
gc.collect(status);
+
EasyMock.verify(context, fs, marker, tserverSet);
}
@@@ -221,24 -183,14 +206,20 @@@
EasyMock.expect(marker.state(server2, id)).andReturn(new Pair<>(WalState.OPEN, path));
EasyMock.replay(context, fs, marker, tserverSet);
- GarbageCollectWriteAheadLogs gc = new GarbageCollectWriteAheadLogs(context, fs, tserverSet) {
- @Override
- protected Map<UUID,Path> getSortedWALogs() {
- return Collections.emptyMap();
- }
-
- @Override
- WalStateManager createWalStateManager(ServerContext serverContext) {
- return marker;
- }
-
- @Override
- Stream<TabletMetadata> createStore(Set<TServerInstance> liveTservers) {
- return tabletOnServer2List;
- }
- };
- var gc =
- new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker, tabletOnServer2List) {
++ GarbageCollectWriteAheadLogs gc =
++ new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker) {
+ @Override
+ protected Map<UUID,Path> getSortedWALogs() {
+ return Collections.emptyMap();
+ }
++
++ @Override
++ Stream<TabletMetadata> createStore(Set<TServerInstance> liveTservers) {
++ return tabletOnServer2List;
++ }
+ };
gc.collect(status);
+
EasyMock.verify(context, fs, marker, tserverSet);
}