You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by "dlmarion (via GitHub)" <gi...@apache.org> on 2023/03/22 19:12:03 UTC
[GitHub] [accumulo] dlmarion opened a new pull request, #3250: Added On-Demand table feature
dlmarion opened a new pull request, #3250:
URL: https://github.com/apache/accumulo/pull/3250
An on-demand table is like an offline table, except that its tablets will be hosted when a client wants to interact with them. A table can be put into an on-demand state using the ondemand shell command or TableOperations.ondemand method. The accumulo.root and accumulo.metadata tables cannot be put into an On-Demand state.
Tablets of an on-demand table are hosted when a client fails to find the tablet location of an on-demand table. When this occurs the client makes an RPC call that inserts an "ondemand" column into the Tablet metadata. In the case that an on-demand tablet needs to be hosted for a client operation, the client will wait for the tablet to be hosted and the tablet location to be resolved by the client before proceeding.
The Manager will assign tablets that have an "ondemand" column in their metadata to TabletServers and will unassign hosted on-demand tablets when the "ondemand" column does not exist. When setting an online table to an on-demand state, all of its tablets will be unloaded. The new Property MANAGER_TABLET_GROUP_WATCHER_INTERVAL specifies the interval at which the Manager will look for tablet state changes. Lower values for this property will reduce the wait time in the client for an on-demand tablet to be hosted.
TabletServers keep track of the last access time for on-demand tablets. Periodically the TabletServer will evaluate which ondemand tablets should be unloaded. The interval for this evaluation is specified by the new Property
TABLE_ONDEMAND_UNLOADER_INTERVAL. At this interval the TabletServer will call the OnDemandTabletUnloader class specified by the Property TABLE_ONDEMAND_UNLOADER, which will return the set of tablets to unload. Unloading is performed by removing the "ondemand" column from the tablet metadata, which will cause the Manager to unassign the tablets. In the case that the TabletServer is running low on memory it will not call the OnDemandTabletUnloader, instead unloading the tablet with the oldest access time.
New metrics are emitted for on-demand tablets. Specifically, the metric accumulo.tserver.tablets.ondemand.unloaded.lowmem is incremented when an on-demand tablet is unloaded for low memory in the TabletServer, and the metric
accumulo.tserver.tablets.ondemand.online is modified when on-demand tablets are hosted or unloaded.
Finally, a new utility, ListOnlineOnDemandTablets, has been created to list on-demand tablets that are currently hosted.
Closes #3210 #3211 #3212
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147425173
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java:
##########
@@ -1479,18 +1463,47 @@ public void online(String tableName)
online(tableName, false);
}
- @Override
- public void online(String tableName, boolean wait)
+ private void changeTableState(String tableName, boolean wait, TableState newState)
throws AccumuloSecurityException, AccumuloException, TableNotFoundException {
EXISTING_TABLE_NAME.validate(tableName);
TableId tableId = context.getTableId(tableName);
+
+ FateOperation op = null;
+ switch (newState) {
+ case OFFLINE:
+ op = FateOperation.TABLE_OFFLINE;
+ if (tableName.equals(MetadataTable.NAME) || tableName.equals(RootTable.NAME)) {
+ throw new AccumuloException("Cannot set table to offline state");
+ }
+ break;
+ case ONDEMAND:
+ op = FateOperation.TABLE_ONDEMAND;
+ if (tableName.equals(MetadataTable.NAME) || tableName.equals(RootTable.NAME)) {
+ throw new AccumuloException("Cannot set table to onDemand state");
+ }
+ break;
+ case ONLINE:
+ op = FateOperation.TABLE_ONLINE;
+ if (tableName.equals(MetadataTable.NAME) || tableName.equals(RootTable.NAME)) {
+ // Don't submit a Fate operation for this, these tables can only be online.
+ return;
+ }
+ break;
+ default:
+ throw new IllegalArgumentException(newState + " is not handled.");
+ }
+
/**
- * ACCUMULO-4574 if table is already online return without executing fate operation.
+ * ACCUMULO-4574 Don't submit a fate operation to change the table state to newState if it's
+ * already in that state.
*/
- if (isOnline(tableName)) {
+ TableState currentState = context.getTableState(tableId, true);
+ if (op == FateOperation.TABLE_ONLINE && currentState == TableState.ONLINE
+ || op == FateOperation.TABLE_ONDEMAND && currentState == TableState.ONDEMAND
+ || op == FateOperation.TABLE_OFFLINE && currentState == TableState.OFFLINE) {
Review Comment:
lol, that's much simpler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147510449
##########
test/src/main/java/org/apache/accumulo/test/functional/OnDemandIT.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.functional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.TabletLocator;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class OnDemandIT extends SharedMiniClusterBase {
Review Comment:
Added in 6409f84
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147512731
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/OnDemandTabletUnloader.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.spi.ondemand;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.TabletId;
+
+/**
+ * Object used by the TabletServer to determine which onDemand Tablets to unload for a Table
+ *
+ * @since 3.1.0
+ */
+public interface OnDemandTabletUnloader {
+
+ interface UnloaderParams {
+
+ /**
+ * @return table configuration
+ * @since 3.1.0
+ */
+ Map<String,String> getTableConfiguration();
+
+ /**
+ * Returns the onDemand tablets that are currently online and the time that they were last
+ * accessed
+ *
+ * @since 3.1.0
+ */
+ Map<TabletId,Long> getOnDemandTablets();
Review Comment:
Changed method name in 6409f84
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147831636
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/DefaultOnDemandTabletUnloader.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.spi.ondemand;
+
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DefaultOnDemandTabletUnloader implements OnDemandTabletUnloader {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultOnDemandTabletUnloader.class);
+ public static final String INACTIVITY_THRESHOLD =
+ "table.custom.ondemand.unloader.inactivity.threshold.seconds";
+ private static final String TEN_MINUTES = Long.toString(MINUTES.toSeconds(10));
+
+ @Override
+ public void evaluate(UnloaderParams params) {
Review Comment:
Unit test added in 8738077. Had to put it down in the tserver module so that I had access to server side objects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#issuecomment-1483126407
Kicked off full IT build
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1150404321
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -210,6 +215,8 @@ public PausedCompactionMetrics getPausedCompactionMetrics() {
private final AtomicLong syncCounter = new AtomicLong(0);
final OnlineTablets onlineTablets = new OnlineTablets();
+ private final Map<KeyExtent,AtomicLong> onDemandTabletAccessTimes =
Review Comment:
Opened #3263 for this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1149207406
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocatorImpl.java:
##########
@@ -524,6 +541,72 @@ public TabletLocation locateTablet(ClientContext context, Text row, boolean skip
}
}
+ @Override
+ public long onDemandTabletsOnlined() {
+ return onDemandTabletsOnlinedCount.get();
+ }
+
+ private void bringOnDemandTabletsOnline(ClientContext context, Range range)
+ throws AccumuloException, AccumuloSecurityException {
+
+ // Confirm that table is in an onDemand state. Don't throw an exception
+ // if the table is not found, calling code will already handle it.
+ try {
+ String tableName = context.getTableName(tableId);
+ if (!context.tableOperations().isOnDemand(tableName)) {
+ log.trace("bringOnDemandTabletsOnline: table {} is not in ondemand state", tableId);
+ return;
+ }
+ } catch (TableNotFoundException e) {
+ log.trace("bringOnDemandTabletsOnline: table not found: {}", tableId);
+ return;
+ }
+
+ final Text scanRangeStart = range.getStartKey().getRow();
+ final Text scanRangeEnd = range.getEndKey().getRow();
+ // Turn the scan range into a KeyExtent and bring online all ondemand tablets
+ // that are overlapped by the scan range
+ final KeyExtent scanRangeKE = new KeyExtent(tableId, scanRangeEnd, scanRangeStart);
+
+ List<TKeyExtent> extentsToBringOnline = new ArrayList<>();
+
+ TabletsMetadata m = context.getAmple().readTablets().forTable(tableId)
+ .overlapping(scanRangeStart, true, null).build();
+ for (TabletMetadata tm : m) {
+ final KeyExtent tabletExtent = tm.getExtent();
+ log.trace("Evaluating tablet {} against range {}", tabletExtent, scanRangeKE);
+ if (tm.getEndRow() != null && tm.getEndRow().compareTo(scanRangeStart) < 0) {
+ // the end row of this tablet is before the start row, skip it
+ log.trace("tablet {} is before scan start range: {}", tabletExtent, scanRangeStart);
+ continue;
+ }
+ if (tm.getPrevEndRow() != null && tm.getPrevEndRow().compareTo(scanRangeEnd) > 0) {
+ // the start row of this tablet is after the scan range end row, skip it
+ log.trace("tablet {} is after scan end range: {}", tabletExtent, scanRangeEnd);
+ continue;
+ }
+ if (scanRangeKE.overlaps(tabletExtent)) {
+ if (!tm.getOnDemand()) {
+ Location loc = tm.getLocation();
+ if (loc != null) {
+ log.debug("tablet {} has location of: {}:{}", loc.getType(), loc.getHostPort());
Review Comment:
Addressed in 85d34f9
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147511121
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsUtil.java:
##########
@@ -101,6 +101,14 @@ public int getMinorCompactionsQueued() {
return result;
}
+ public int getOnDemandOnlineCount() {
+ return tserver.getOnDemandOnlineCount();
Review Comment:
Updated in 6409f84 to use the set of online tablets. I removed the counter and the code that referenced it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1145394043
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
Resolved in 83bc60a
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
Resolved in 83bc60a
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147514717
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java:
##########
@@ -1479,18 +1463,47 @@ public void online(String tableName)
online(tableName, false);
}
- @Override
- public void online(String tableName, boolean wait)
+ private void changeTableState(String tableName, boolean wait, TableState newState)
throws AccumuloSecurityException, AccumuloException, TableNotFoundException {
EXISTING_TABLE_NAME.validate(tableName);
TableId tableId = context.getTableId(tableName);
+
+ FateOperation op = null;
+ switch (newState) {
+ case OFFLINE:
+ op = FateOperation.TABLE_OFFLINE;
+ if (tableName.equals(MetadataTable.NAME) || tableName.equals(RootTable.NAME)) {
+ throw new AccumuloException("Cannot set table to offline state");
+ }
+ break;
+ case ONDEMAND:
+ op = FateOperation.TABLE_ONDEMAND;
+ if (tableName.equals(MetadataTable.NAME) || tableName.equals(RootTable.NAME)) {
+ throw new AccumuloException("Cannot set table to onDemand state");
+ }
+ break;
+ case ONLINE:
+ op = FateOperation.TABLE_ONLINE;
+ if (tableName.equals(MetadataTable.NAME) || tableName.equals(RootTable.NAME)) {
+ // Don't submit a Fate operation for this, these tables can only be online.
+ return;
+ }
+ break;
+ default:
+ throw new IllegalArgumentException(newState + " is not handled.");
+ }
+
/**
- * ACCUMULO-4574 if table is already online return without executing fate operation.
+ * ACCUMULO-4574 Don't submit a fate operation to change the table state to newState if it's
+ * already in that state.
*/
- if (isOnline(tableName)) {
+ TableState currentState = context.getTableState(tableId, true);
+ if (op == FateOperation.TABLE_ONLINE && currentState == TableState.ONLINE
+ || op == FateOperation.TABLE_ONDEMAND && currentState == TableState.ONDEMAND
+ || op == FateOperation.TABLE_OFFLINE && currentState == TableState.OFFLINE) {
Review Comment:
Modified in 6409f84
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147511505
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -760,6 +769,12 @@ public void run() {
}
final AccumuloConfiguration aconf = getConfiguration();
+ final long onDemandUnloaderInterval =
+ aconf.getTimeInMillis(Property.TABLE_ONDEMAND_UNLOADER_INTERVAL);
Review Comment:
Updated in 6409f84 to be a TSERV property
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147830611
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,283 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
Review Comment:
Addressed in 8738077
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,283 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
Review Comment:
Addressed in 8738077
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147832528
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/OnDemandTabletUnloader.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.spi.ondemand;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.TabletId;
+
+/**
+ * Object used by the TabletServer to determine which onDemand Tablets to unload for a Table
+ *
+ * @since 3.1.0
+ */
+public interface OnDemandTabletUnloader {
+
+ interface UnloaderParams {
+
+ /**
+ * @return table configuration
+ * @since 3.1.0
+ */
+ Map<String,String> getTableConfiguration();
Review Comment:
Addressed in 8738077
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#issuecomment-1480123717
If you want to test this yourself, you can:
1. Build Accumulo from this branch
2. Set the following properties in accumulo.properties:
```
manager.tablet.watcher.interval=15s
table.ondemand.tablet.unloader.interval=1m
table.custom.ondemand.unloader.inactivity.threshold=120000
```
3. Start up local instance
4. Log into Accumulo shell, and do the following
```
createtable test
```
> **Note**
> notice test and 1 tablet in Monitor
> notice "loc" column in tablet metadata
```
ondemand test
```
> **Note**
> notice 0 tablets in monitor
> notice "loc" missing from tablet metadata
```
insert a b c d
```
> **Note**
> notice shell command wait for tablet to be hosted
> notice 1 tablet in Monitor
> notice "loc" column in tablet metadata
> notice "ondemand" column in tablet metadata
```
addsplits m -t test
```
> **Note**
> notice 1 tablet in Monitor
> notice "loc" in tablet metadata for source tablet only
> notice "ondemand" column in tablet metadata for source tablet only
```
sleep 240
```
> **Note**
> table.custom.ondemand.unloader.inactivity.threshold is set to 2m, waiting 4m should unload the tablet
> notice 0 tablets in monitor
> notice "loc" column missing in tablet metadata
> notice "ondemand" column missing in tablet metadata
```
insert a b c d
```
> **Note**
> notice shell command wait for tablet to be hosted
> notice 1 tablet in Monitor
> notice "loc" column in tablet metadata
> notice "ondemand" column in tablet metadata
5. Stop Accumulo
6. Start Accumulo
> **Note**
> test tablet had "ondemand" column on shutdown, so it gets re-hosted
> notice 1 tablet in Monitor
> notice "loc" column in tablet metadata
> notice "ondemand" column in tablet metadata
7. Wait a few minutes for tablet to be unloaded
> **Note**
> notice 0 tablets in monitor
> notice "loc" column missing in tablet metadata
> notice "ondemand" column missing in tablet metadata
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1150405414
##########
server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/LoadFiles.java:
##########
@@ -107,6 +107,7 @@ public long isReady(long tid, Manager manager) throws Exception {
@Override
public Repo<Manager> call(final long tid, final Manager manager) {
+ // TODO: How are we treating ONDEMAND tables for BulkImport?
Review Comment:
Opened #3264 for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1149208341
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java:
##########
@@ -1555,4 +1556,21 @@ public TSummaries contiuneGetSummaries(TInfo tinfo, long sessionId)
return handleTimeout(sessionId);
}
}
+
+ @Override
+ public void bringOnDemandTabletsOnline(TInfo tinfo, TCredentials credentials, String tableId,
+ List<TKeyExtent> extents) throws ThriftSecurityException, TException {
+ final TableId tid = TableId.of(tableId);
+ NamespaceId namespaceId = getNamespaceId(credentials, tid);
+ if (!security.canScan(credentials, tid, namespaceId)) {
+ throw new ThriftSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED);
+ }
+ try (TabletsMutator mutator = this.context.getAmple().mutateTablets()) {
+ extents.forEach(e -> {
+ mutator.mutateTablet(KeyExtent.fromThrift(e)).putOnDemand().mutate();
+ });
Review Comment:
Addressed in 85d34f9
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147832046
##########
test/src/main/java/org/apache/accumulo/test/functional/OnDemandIT.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.functional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+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.AccumuloSecurityException;
+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.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.TabletLocator;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import com.google.common.collect.Iterables;
+
+public class OnDemandIT extends SharedMiniClusterBase {
+
+ @FunctionalInterface
+ public static interface Action {
+ void execute(AccumuloClient c, String tableName);
+ }
+
+ private static final int managerTabletGroupWatcherInterval = 5;
+ private static final int inactiveOnDemandTabletUnloaderInterval = 10;
+ private static TestStatsDSink sink;
+ private static Thread metricConsumer;
+ private static Long ONDEMAND_ONLINE_COUNT = 0L;
+
+ @BeforeAll
+ public static void beforeAll() throws Exception {
+ sink = new TestStatsDSink();
+ metricConsumer = new Thread(() -> {
+ while (!Thread.currentThread().isInterrupted()) {
+ List<String> statsDMetrics = sink.getLines();
+ for (String line : statsDMetrics) {
+ if (Thread.currentThread().isInterrupted()) {
+ break;
+ }
+ if (line.startsWith("accumulo")) {
+ Metric metric = TestStatsDSink.parseStatsDMetric(line);
+ if (MetricsProducer.METRICS_TSERVER_TABLETS_ONLINE_ONDEMAND.equals(metric.getName())) {
+ Long val = Long.parseLong(metric.getValue());
+ ONDEMAND_ONLINE_COUNT = val;
+ }
+ }
+ }
+ }
+ });
+ metricConsumer.start();
+ SharedMiniClusterBase.startMiniClusterWithConfig((cfg, core) -> {
+ cfg.setNumTservers(1);
+ cfg.setProperty(Property.MANAGER_TABLET_GROUP_WATCHER_INTERVAL,
+ Integer.toString(managerTabletGroupWatcherInterval));
+ cfg.setProperty(Property.TSERV_ONDEMAND_UNLOADER_INTERVAL,
+ Integer.toString(inactiveOnDemandTabletUnloaderInterval));
+ cfg.setProperty("table.custom.ondemand.unloader.inactivity.threshold.seconds", "15");
+
+ // Tell the server processes to use a StatsDMeterRegistry that will be configured
+ // to push all metrics to the sink we started.
+ cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+ cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY,
+ TestStatsDRegistryFactory.class.getName());
+ Map<String,String> sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1",
+ TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort()));
+ cfg.setSystemProperties(sysProps);
+ });
+ }
+
+ @AfterAll
+ public static void after() throws Exception {
+ sink.close();
+ metricConsumer.interrupt();
+ metricConsumer.join();
+ }
+
+ @BeforeEach
+ public void before() {
+ ONDEMAND_ONLINE_COUNT = 0L;
+ }
+
+ public void testLoadUnload(Action action) throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ ManagerAssignmentIT.loadDataForScan(c, tableName);
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(2 * managerTabletGroupWatcherInterval * 1000);
+
+ List<TabletStats> stats = ManagerAssignmentIT.getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+ assertEquals(0, ONDEMAND_ONLINE_COUNT);
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ // execute the action that will cause the tablets to be brought online
+ action.execute(c, tableName);
+
+ while (ONDEMAND_ONLINE_COUNT != 4) {
+ Thread.sleep(500);
+ }
+ stats = ManagerAssignmentIT.getTabletStats(c, tableId);
+ assertEquals(4, stats.size());
+ assertEquals(4, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // wait for the inactivity timeout to cause the tablets to be unloaded from the
+ // TabletServer
+ while (ONDEMAND_ONLINE_COUNT != 0) {
+ Thread.sleep(500);
+ }
+ stats = ManagerAssignmentIT.getTabletStats(c, tableId);
+ assertEquals(0, stats.size());
+
+ }
+ }
+
+ @Test
+ public void testLoadUnloadUsingWriter() throws Exception {
+ testLoadUnload((client, t) -> {
+ try {
+ // load the same data again, this will cause the tablets to be brought online.
+ ManagerAssignmentIT.loadDataForScan(client, t);
+ } catch (MutationsRejectedException | TableNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ @Test
+ public void testLoadUnloadUsingScan() throws Exception {
+ testLoadUnload((client, t) -> {
+ try {
+ // scan the table, this will cause the tablets to be brought online.
+ Scanner s = client.createScanner(t);
+ Iterables.size(s);
Review Comment:
Addressed in 8738077
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/OnDemandTabletUnloader.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.spi.ondemand;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.TabletId;
+
+/**
+ * Object used by the TabletServer to determine which onDemand Tablets to unload for a Table
+ *
+ * @since 3.1.0
+ */
+public interface OnDemandTabletUnloader {
+
+ interface UnloaderParams {
+
+ /**
+ * @return table configuration
+ * @since 3.1.0
+ */
+ Map<String,String> getTableConfiguration();
+
+ /**
+ * Returns the onDemand tablets that are currently online and the time that they were last
+ * accessed. Access times are in nanoseconds, from {@link System#nanoTime()}, and should be
+ * handled accordingly.
+ *
+ * @since 3.1.0
+ */
+ Map<TabletId,Long> getLastAccessTimes();
+
+ /**
+ * Called by the implementation to inform the TabletServer as to which onDemand tablets should
+ * be unloaded
Review Comment:
Addressed in 8738077
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147592211
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/OnDemandTabletUnloader.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.spi.ondemand;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.TabletId;
+
+/**
+ * Object used by the TabletServer to determine which onDemand Tablets to unload for a Table
+ *
+ * @since 3.1.0
+ */
+public interface OnDemandTabletUnloader {
+
+ interface UnloaderParams {
+
+ /**
+ * @return table configuration
+ * @since 3.1.0
+ */
+ Map<String,String> getTableConfiguration();
+
+ /**
+ * Returns the onDemand tablets that are currently online and the time that they were last
+ * accessed. Access times are in nanoseconds, from {@link System#nanoTime()}, and should be
+ * handled accordingly.
+ *
+ * @since 3.1.0
+ */
+ Map<TabletId,Long> getLastAccessTimes();
+
+ /**
+ * Called by the implementation to inform the TabletServer as to which onDemand tablets should
+ * be unloaded
Review Comment:
Could document expectations when there is nothing to unload.
```suggestion
* Called by the implementation to inform the TabletServer as to which onDemand tablets should
* be unloaded. When nothing is found to unload its ok to pass in an empty set or not call this method.
```
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/OnDemandTabletUnloader.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.spi.ondemand;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.TabletId;
+
+/**
+ * Object used by the TabletServer to determine which onDemand Tablets to unload for a Table
+ *
+ * @since 3.1.0
+ */
+public interface OnDemandTabletUnloader {
+
+ interface UnloaderParams {
+
+ /**
+ * @return table configuration
+ * @since 3.1.0
+ */
+ Map<String,String> getTableConfiguration();
Review Comment:
This could be offer the [ServiceEnvironment](https://github.com/apache/accumulo/blob/main/core/src/main/java/org/apache/accumulo/core/spi/common/ServiceEnvironment.java) which exposes table config among other things. I think the exposed table config has convenience methods for getting table custom props also.
```suggestion
ServiceEnvironment getServiceEnvironment();
```
##########
core/src/main/java/org/apache/accumulo/core/conf/Property.java:
##########
@@ -309,6 +309,9 @@ public enum Property {
"The balancer class that accumulo will use to make tablet assignment and "
+ "migration decisions.",
"1.3.5"),
+ MANAGER_TABLET_GROUP_WATCHER_INTERVAL("manager.tablet.watcher.interval", "60s",
+ PropertyType.TIMEDURATION,
+ "Time to wait between scanning tablet states to determine migrations, etc.", "3.1.0"),
Review Comment:
Could make a PR for the 3 branch to add this change. Could still include the change here, we can make the change in main branch and when we merge main into elastic branch just resolve any conflicts. If we get this change in main then it will be one less divergence to worry about as we merge main into elastic branch going forward.
##########
server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletStateChangeIterator.java:
##########
@@ -191,26 +194,29 @@ protected void consume() throws IOException {
}
// is the table supposed to be online or offline?
- boolean shouldBeOnline = onlineTables.contains(tls.extent.tableId());
+ final boolean shouldBeOnline = onlineTables.contains(tls.extent.tableId());
+ final boolean isOnDemandTable = onDemandTables.contains(tls.extent.tableId());
+ final boolean onDemandTabletShouldBeHosted = tls.ondemand;
Review Comment:
May be able to compute everything here and then later in the switch stmt only use shouldBeOnline var.
```suggestion
final boolean shouldBeOnline = onlineTables.contains(tls.extent.tableId()) || (onDemandTables.contains(tls.extent.tableId()) && tls.ondemand);
```
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,283 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
Review Comment:
while loop if possible
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,283 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
Review Comment:
Instead of sleeping a fixed time could this do
```
while(countHostedTablets() > 0) {
sleep(50);
}
```
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,283 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
Review Comment:
Could do a while loop here if possible too.
##########
test/src/main/java/org/apache/accumulo/test/functional/OnDemandIT.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.functional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+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.AccumuloSecurityException;
+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.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.TabletLocator;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import com.google.common.collect.Iterables;
+
+public class OnDemandIT extends SharedMiniClusterBase {
+
+ @FunctionalInterface
+ public static interface Action {
+ void execute(AccumuloClient c, String tableName);
+ }
+
+ private static final int managerTabletGroupWatcherInterval = 5;
+ private static final int inactiveOnDemandTabletUnloaderInterval = 10;
+ private static TestStatsDSink sink;
+ private static Thread metricConsumer;
+ private static Long ONDEMAND_ONLINE_COUNT = 0L;
+
+ @BeforeAll
+ public static void beforeAll() throws Exception {
+ sink = new TestStatsDSink();
+ metricConsumer = new Thread(() -> {
+ while (!Thread.currentThread().isInterrupted()) {
+ List<String> statsDMetrics = sink.getLines();
+ for (String line : statsDMetrics) {
+ if (Thread.currentThread().isInterrupted()) {
+ break;
+ }
+ if (line.startsWith("accumulo")) {
+ Metric metric = TestStatsDSink.parseStatsDMetric(line);
+ if (MetricsProducer.METRICS_TSERVER_TABLETS_ONLINE_ONDEMAND.equals(metric.getName())) {
+ Long val = Long.parseLong(metric.getValue());
+ ONDEMAND_ONLINE_COUNT = val;
+ }
+ }
+ }
+ }
+ });
+ metricConsumer.start();
+ SharedMiniClusterBase.startMiniClusterWithConfig((cfg, core) -> {
+ cfg.setNumTservers(1);
+ cfg.setProperty(Property.MANAGER_TABLET_GROUP_WATCHER_INTERVAL,
+ Integer.toString(managerTabletGroupWatcherInterval));
+ cfg.setProperty(Property.TSERV_ONDEMAND_UNLOADER_INTERVAL,
+ Integer.toString(inactiveOnDemandTabletUnloaderInterval));
+ cfg.setProperty("table.custom.ondemand.unloader.inactivity.threshold.seconds", "15");
+
+ // Tell the server processes to use a StatsDMeterRegistry that will be configured
+ // to push all metrics to the sink we started.
+ cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+ cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY,
+ TestStatsDRegistryFactory.class.getName());
+ Map<String,String> sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1",
+ TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort()));
+ cfg.setSystemProperties(sysProps);
+ });
+ }
+
+ @AfterAll
+ public static void after() throws Exception {
+ sink.close();
+ metricConsumer.interrupt();
+ metricConsumer.join();
+ }
+
+ @BeforeEach
+ public void before() {
+ ONDEMAND_ONLINE_COUNT = 0L;
+ }
+
+ public void testLoadUnload(Action action) throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ ManagerAssignmentIT.loadDataForScan(c, tableName);
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(2 * managerTabletGroupWatcherInterval * 1000);
+
+ List<TabletStats> stats = ManagerAssignmentIT.getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+ assertEquals(0, ONDEMAND_ONLINE_COUNT);
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ // execute the action that will cause the tablets to be brought online
+ action.execute(c, tableName);
+
+ while (ONDEMAND_ONLINE_COUNT != 4) {
+ Thread.sleep(500);
+ }
+ stats = ManagerAssignmentIT.getTabletStats(c, tableId);
+ assertEquals(4, stats.size());
+ assertEquals(4, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // wait for the inactivity timeout to cause the tablets to be unloaded from the
+ // TabletServer
+ while (ONDEMAND_ONLINE_COUNT != 0) {
+ Thread.sleep(500);
+ }
+ stats = ManagerAssignmentIT.getTabletStats(c, tableId);
+ assertEquals(0, stats.size());
+
+ }
+ }
+
+ @Test
+ public void testLoadUnloadUsingWriter() throws Exception {
+ testLoadUnload((client, t) -> {
+ try {
+ // load the same data again, this will cause the tablets to be brought online.
+ ManagerAssignmentIT.loadDataForScan(client, t);
+ } catch (MutationsRejectedException | TableNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ @Test
+ public void testLoadUnloadUsingScan() throws Exception {
+ testLoadUnload((client, t) -> {
+ try {
+ // scan the table, this will cause the tablets to be brought online.
+ Scanner s = client.createScanner(t);
+ Iterables.size(s);
Review Comment:
Would be good to validate data coming from the scanner is as expected in the test.
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java:
##########
@@ -1479,18 +1463,45 @@ public void online(String tableName)
online(tableName, false);
}
- @Override
- public void online(String tableName, boolean wait)
+ private void changeTableState(String tableName, boolean wait, TableState newState)
Review Comment:
I like how this change dedupes code.
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/DefaultOnDemandTabletUnloader.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.spi.ondemand;
+
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DefaultOnDemandTabletUnloader implements OnDemandTabletUnloader {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultOnDemandTabletUnloader.class);
+ public static final String INACTIVITY_THRESHOLD =
+ "table.custom.ondemand.unloader.inactivity.threshold.seconds";
+ private static final String TEN_MINUTES = Long.toString(MINUTES.toSeconds(10));
+
+ @Override
+ public void evaluate(UnloaderParams params) {
Review Comment:
Could add a simple unit test for this class. One way to do this would be to have a protected getTime method that the unit test overrides. If this were done it could be in a follow on PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] cshannon commented on pull request #3250: Added On-Demand table feature
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#issuecomment-1483896573
My latest changes from #3257 caused merged conflicts in this branch since I changed things in tests and especially TabletLocationState. I went ahead and created a PR for this branch that merges in elasticity branch and fixes all the conflicts: https://github.com/dlmarion/accumulo/pull/39
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1149207717
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -1285,4 +1304,153 @@ public BlockCacheConfiguration getBlockCacheConfiguration(AccumuloConfiguration
return BlockCacheConfiguration.forTabletServer(acuConf);
}
+ public int getOnDemandOnlineUnloadedForLowMemory() {
+ return onDemandUnloadedLowMemory.get();
+ }
+
+ // called from AssignmentHandler
+ public void insertOnDemandAccessTime(KeyExtent extent) {
+ onDemandTabletAccessTimes.putIfAbsent(extent, new AtomicLong(System.nanoTime()));
+ }
+
+ // called from getOnlineExtent
+ private void updateOnDemandAccessTime(KeyExtent extent) {
+ final long currentTime = System.nanoTime();
+ AtomicLong l = onDemandTabletAccessTimes.get(extent);
+ if (l != null) {
+ l.set(currentTime);
+ }
+ }
+
+ // called from UnloadTabletHandler
+ public void removeOnDemandAccessTime(KeyExtent extent) {
+ onDemandTabletAccessTimes.remove(extent);
+ }
+
+ private boolean isTabletInUse(KeyExtent extent) {
+ // Don't call getOnlineTablet as that will update the last access time
+ final Tablet t = onlineTablets.snapshot().get(extent);
+ if (t == null) {
+ return false;
+ }
+ t.updateRates(System.currentTimeMillis());
+ if (t.ingestRate() != 0.0 && t.queryRate() != 0.0 && t.scanRate() != 0.0) {
+ // tablet is ingesting or scanning
+ return true;
+ }
+ return false;
+ }
+
+ public void evaluateOnDemandTabletsForUnload() {
+
+ final SortedMap<KeyExtent,Tablet> online = getOnlineTablets();
+
+ // Find and remove access time entries for KeyExtents
+ // that are no longer in the onlineTablets collection
+ Set<KeyExtent> missing = onDemandTabletAccessTimes.keySet().stream()
+ .filter(k -> !online.containsKey(k)).collect(Collectors.toSet());
+ if (!missing.isEmpty()) {
+ log.debug("Removing onDemandAccessTimes for tablets as tablets no longer online: {}",
+ missing);
+ missing.forEach(onDemandTabletAccessTimes::remove);
+ if (onDemandTabletAccessTimes.isEmpty()) {
+ return;
+ }
+ }
+
+ // It's possible, from a tablet split or merge for example,
+ // that there is an onDemand tablet that is hosted for which
+ // we have no access time. Add any missing online onDemand
+ // tablets
+ online.forEach((k, v) -> {
+ if (v.isOnDemand() && !onDemandTabletAccessTimes.containsKey(k)) {
+ insertOnDemandAccessTime(k);
+ }
+ });
+
+ log.debug("Evaluating online onDemand tablets: {}", onDemandTabletAccessTimes);
+
+ if (onDemandTabletAccessTimes.isEmpty()) {
+ return;
+ }
+
+ // If the TabletServer is running low on memory, don't call the SPI
+ // plugin to evaluate which onDemand tablets to unload, just get the
+ // onDemand tablet with the oldest access time and unload it.
+ if (getContext().getLowMemoryDetector().isRunningLowOnMemory()) {
+ final SortedMap<Long,KeyExtent> timeSortedOnDemandExtents = new TreeMap<>();
+ onDemandTabletAccessTimes.forEach((k, v) -> timeSortedOnDemandExtents.put(v.get(), k));
+ Long oldestAccessTime = timeSortedOnDemandExtents.firstKey();
+ KeyExtent oldestKeyExtent = timeSortedOnDemandExtents.get(oldestAccessTime);
+ log.warn("Unloading onDemand tablet: {} for table: {} due to low memory", oldestKeyExtent,
+ oldestKeyExtent.tableId());
+ getContext().getAmple().mutateTablet(oldestKeyExtent).deleteOnDemand().mutate();
+ onDemandUnloadedLowMemory.addAndGet(1);
+ return;
+ }
+
+ // onDemandTabletAccessTimes is a HashMap. Sort the extents
+ // so that we can process them by table.
+ final SortedMap<KeyExtent,AtomicLong> sortedOnDemandExtents =
+ new TreeMap<KeyExtent,AtomicLong>();
+ sortedOnDemandExtents.putAll(onDemandTabletAccessTimes);
+
+ // The access times are updated when getOnlineTablet is called by other methods,
+ // but may not necessarily capture whether or not the Tablet is currently being used.
+ // For example, getOnlineTablet is called from startScan but not from continueScan.
+ // Instead of instrumenting all of the locations where the tablet is touched we
+ // can use the Tablet metrics.
+ final Set<KeyExtent> onDemandTabletsInUse = new HashSet<>();
+ for (KeyExtent extent : sortedOnDemandExtents.keySet()) {
+ if (isTabletInUse(extent)) {
+ onDemandTabletsInUse.add(extent);
+ }
+ }
+ if (!onDemandTabletsInUse.isEmpty()) {
+ log.debug("Removing onDemandAccessTimes for tablets as tablets are in use: {}",
+ onDemandTabletsInUse);
+ onDemandTabletsInUse.forEach(sortedOnDemandExtents::remove);
+ if (sortedOnDemandExtents.isEmpty()) {
+ return;
+ }
+ }
+
+ Set<TableId> tableIds = sortedOnDemandExtents.keySet().stream().map((k) -> {
+ return k.tableId();
+ }).distinct().collect(Collectors.toSet());
+ log.debug("Tables that have online onDemand tablets: {}", tableIds);
+ final Map<TableId,OnDemandTabletUnloader> unloaders = new HashMap<>();
+ tableIds.forEach(tid -> {
+ TableConfiguration tconf = getContext().getTableConfiguration(tid);
+ String tableContext = ClassLoaderUtil.tableContext(tconf);
+ String unloaderClassName = tconf.get(Property.TABLE_ONDEMAND_UNLOADER);
+ try {
+ Class<? extends OnDemandTabletUnloader> clazz = ClassLoaderUtil.loadClass(tableContext,
+ unloaderClassName, OnDemandTabletUnloader.class);
+ unloaders.put(tid, clazz.getConstructor().newInstance());
+ } catch (ClassNotFoundException | InstantiationException | IllegalAccessException
+ | IllegalArgumentException | InvocationTargetException | NoSuchMethodException
+ | SecurityException e) {
+ log.error(
+ "Error constructing OnDemandTabletUnloader implementation, not unloading onDemand tablets",
+ e);
+ return;
+ }
+ });
+ tableIds.forEach(tid -> {
+ Map<KeyExtent,AtomicLong> subset = sortedOnDemandExtents.entrySet().stream().filter((e) -> {
+ return e.getKey().tableId().equals(tid);
+ }).collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue()));
Review Comment:
Addressed in 85d34f9
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1146173294
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java:
##########
@@ -1479,18 +1463,47 @@ public void online(String tableName)
online(tableName, false);
}
- @Override
- public void online(String tableName, boolean wait)
+ private void changeTableState(String tableName, boolean wait, TableState newState)
throws AccumuloSecurityException, AccumuloException, TableNotFoundException {
EXISTING_TABLE_NAME.validate(tableName);
TableId tableId = context.getTableId(tableName);
+
+ FateOperation op = null;
+ switch (newState) {
+ case OFFLINE:
+ op = FateOperation.TABLE_OFFLINE;
+ if (tableName.equals(MetadataTable.NAME) || tableName.equals(RootTable.NAME)) {
+ throw new AccumuloException("Cannot set table to offline state");
+ }
+ break;
+ case ONDEMAND:
+ op = FateOperation.TABLE_ONDEMAND;
+ if (tableName.equals(MetadataTable.NAME) || tableName.equals(RootTable.NAME)) {
+ throw new AccumuloException("Cannot set table to onDemand state");
+ }
+ break;
+ case ONLINE:
+ op = FateOperation.TABLE_ONLINE;
+ if (tableName.equals(MetadataTable.NAME) || tableName.equals(RootTable.NAME)) {
+ // Don't submit a Fate operation for this, these tables can only be online.
+ return;
+ }
+ break;
+ default:
+ throw new IllegalArgumentException(newState + " is not handled.");
+ }
+
/**
- * ACCUMULO-4574 if table is already online return without executing fate operation.
+ * ACCUMULO-4574 Don't submit a fate operation to change the table state to newState if it's
+ * already in that state.
*/
- if (isOnline(tableName)) {
+ TableState currentState = context.getTableState(tableId, true);
+ if (op == FateOperation.TABLE_ONLINE && currentState == TableState.ONLINE
+ || op == FateOperation.TABLE_ONDEMAND && currentState == TableState.ONDEMAND
+ || op == FateOperation.TABLE_OFFLINE && currentState == TableState.OFFLINE) {
Review Comment:
```suggestion
if (newState == currentState) {
```
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/OnDemandTabletUnloader.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.spi.ondemand;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.TabletId;
+
+/**
+ * Object used by the TabletServer to determine which onDemand Tablets to unload for a Table
+ *
+ * @since 3.1.0
+ */
+public interface OnDemandTabletUnloader {
+
+ interface UnloaderParams {
+
+ /**
+ * @return table configuration
+ * @since 3.1.0
+ */
+ Map<String,String> getTableConfiguration();
+
+ /**
+ * Returns the onDemand tablets that are currently online and the time that they were last
+ * accessed
+ *
+ * @since 3.1.0
+ */
+ Map<TabletId,Long> getOnDemandTablets();
Review Comment:
In the future other stats/info may be offered. I would recommend a more specific method name.
```suggestion
Map<TabletId,Long> getLastAccessTimes();
```
##########
server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java:
##########
@@ -178,6 +188,13 @@ public synchronized void transitionTableState(final TableId tableId, final Table
log.error("FATAL Failed to transition table to state {}", newState);
throw new RuntimeException(e);
}
+ // Remove onDemand columns from all tablets
+ if (tableId != RootTable.ID && tableId != MetadataTable.ID
+ && (newState == TableState.ONLINE || newState == TableState.OFFLINE)) {
+ this.context.getAmple().readTablets().forTable(tableId).build().forEach(tm -> {
+ this.context.getAmple().mutateTablet(tm.getExtent()).deleteOnDemand().mutate();
+ });
Review Comment:
The following code will create a single batch writer and use it for all updates. Maybe it will compile. I think the current code will create and flush a batch writer per mutation.
```suggestion
try(var tabletsMutator = context.getAmple().mutateTablets()){
this.context.getAmple().readTablets().forTable(tableId).build().forEach(tm -> {
tabletsMutator.mutateTablet(tm.getExtent()).deleteOnDemand().mutate();
});
}
```
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/OnDemandTabletUnloader.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.spi.ondemand;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.TabletId;
+
+/**
+ * Object used by the TabletServer to determine which onDemand Tablets to unload for a Table
+ *
+ * @since 3.1.0
+ */
+public interface OnDemandTabletUnloader {
+
+ interface UnloaderParams {
+
+ /**
+ * @return table configuration
+ * @since 3.1.0
+ */
+ Map<String,String> getTableConfiguration();
+
+ /**
+ * Returns the onDemand tablets that are currently online and the time that they were last
+ * accessed
Review Comment:
Need to specifiy in the javadoc how this time was obtained so user can reason about it. i.e. was is it from System.currentTimeMillis or System.nanoTime.
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -210,6 +215,8 @@ public PausedCompactionMetrics getPausedCompactionMetrics() {
private final AtomicLong syncCounter = new AtomicLong(0);
final OnlineTablets onlineTablets = new OnlineTablets();
+ private final Map<KeyExtent,AtomicLong> onDemandTabletAccessTimes =
Review Comment:
Wondering if it would simplify things if the last access time were stored in the tablet object. Then this map does not need to be maintained and have the possibility to drift from the set of online tablets. When this map is updated could instead call a method on the tablet.
Also wondering if it would be better for the tablet code to take ownership of updating the last access time rather than the tserver code. Tserver code could read it and tablet code update it as needed.
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -760,6 +769,12 @@ public void run() {
}
final AccumuloConfiguration aconf = getConfiguration();
+ final long onDemandUnloaderInterval =
+ aconf.getTimeInMillis(Property.TABLE_ONDEMAND_UNLOADER_INTERVAL);
Review Comment:
This is a per table prop but it seems like this is running at the tserver level, so if different tables have different intervals then not sure that would have any impact.
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java:
##########
@@ -1716,9 +1760,9 @@ public void exportTable(String tableName, String exportDir)
EXISTING_TABLE_NAME.validate(tableName);
checkArgument(exportDir != null, "exportDir is null");
- if (isOnline(tableName)) {
- throw new IllegalStateException("The table " + tableName + " is online; exportTable requires"
- + " a table to be offline before exporting.");
+ if (isOnline(tableName) || isOnDemand(tableName)) {
+ throw new IllegalStateException("The table " + tableName
+ + " is not offline; exportTable requires a table to be offline before exporting.");
Review Comment:
This comment made me realize investigation may be needed. Created the following item in the elasticity project.
https://github.com/orgs/apache/projects/164/views/1?pane=issue&itemId=23727312
##########
test/src/main/java/org/apache/accumulo/test/functional/OnDemandIT.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.functional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.TabletLocator;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class OnDemandIT extends SharedMiniClusterBase {
Review Comment:
We need test for scanning, but I don't think we can fully do that yet. I opened #3252 and #3253 as follow on work for this.
##########
server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/LoadFiles.java:
##########
@@ -107,6 +107,7 @@ public long isReady(long tid, Manager manager) throws Exception {
@Override
public Repo<Manager> call(final long tid, final Manager manager) {
+ // TODO: How are we treating ONDEMAND tables for BulkImport?
Review Comment:
In V2 bulk import when we switch it to use conditional mutations that may completely address this.
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerMetricsUtil.java:
##########
@@ -101,6 +101,14 @@ public int getMinorCompactionsQueued() {
return result;
}
+ public int getOnDemandOnlineCount() {
+ return tserver.getOnDemandOnlineCount();
Review Comment:
Other code in this class iterates over all of the online tablets to get the counts. If this code did that then counter would not need to maintained (which can be tricky to keep correct, especially as the code changes). Also it would make this code follow the pattern of the other metrics. If this pattern of iterating over all the tablet is not good then it would probably be better to consider improving it for all metrics and not optimize a single metric in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1145392989
##########
test/src/main/java/org/apache/accumulo/test/functional/OnDemandIT.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.functional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.TabletLocator;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class OnDemandIT extends SharedMiniClusterBase {
+
+ private static final int managerTabletGroupWatcherInterval = 5;
+ private static final int inactiveOnDemandTabletUnloaderInterval = 30;
+ private static TestStatsDSink sink;
+ private static Thread metricConsumer;
+ private static Long ONDEMAND_ONLINE_COUNT = 0L;
+
+ @BeforeAll
+ public static void beforeAll() throws Exception {
+ sink = new TestStatsDSink();
+ metricConsumer = new Thread(() -> {
+ while (!Thread.currentThread().isInterrupted()) {
+ List<String> statsDMetrics = sink.getLines();
+ for (String line : statsDMetrics) {
+ if (Thread.currentThread().isInterrupted()) {
+ break;
+ }
+ if (line.startsWith("accumulo")) {
+ Metric metric = TestStatsDSink.parseStatsDMetric(line);
+ if (MetricsProducer.METRICS_TSERVER_TABLETS_ONLINE_ONDEMAND.equals(metric.getName())) {
+ Long val = Long.parseLong(metric.getValue());
+ ONDEMAND_ONLINE_COUNT = val;
+ }
+ }
+ }
+ }
+ });
+ metricConsumer.start();
+ SharedMiniClusterBase.startMiniClusterWithConfig((cfg, core) -> {
+ cfg.setNumTservers(1);
+ cfg.setProperty(Property.MANAGER_TABLET_GROUP_WATCHER_INTERVAL,
+ Integer.toString(managerTabletGroupWatcherInterval));
+ cfg.setProperty(Property.TABLE_ONDEMAND_UNLOADER_INTERVAL,
+ Integer.toString(inactiveOnDemandTabletUnloaderInterval));
+ cfg.setProperty("table.custom.ondemand.unloader.inactivity.threshold", "30000");
+
+ // Tell the server processes to use a StatsDMeterRegistry that will be configured
+ // to push all metrics to the sink we started.
+ cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+ cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY,
+ TestStatsDRegistryFactory.class.getName());
+ Map<String,String> sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1",
+ TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort()));
+ cfg.setSystemProperties(sysProps);
+ });
+ }
+
+ @AfterAll
+ public static void after() throws Exception {
+ sink.close();
+ metricConsumer.interrupt();
+ metricConsumer.join();
+ }
+
+ @BeforeEach
+ public void before() {
+ ONDEMAND_ONLINE_COUNT = 0L;
+ }
+
+ @Test
+ public void testLoadUnload() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ ManagerAssignmentIT.loadDataForScan(c, tableName);
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
Resolved in 83bc60a
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1145394350
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
Resolved in 83bc60a
##########
test/src/main/java/org/apache/accumulo/test/MultiTableBatchWriterIT.java:
##########
@@ -400,4 +401,56 @@ public void testOfflineTable() throws Exception {
assertTrue(mutationsRejected, "Expected mutations to be rejected.");
}
+
+ @Test
+ public void testOnDemandTable() throws Exception {
+ boolean mutationsRejected = false;
+
+ try {
+ final String[] names = getUniqueNames(2);
+ final String table1 = names[0], table2 = names[1];
+
+ TableOperations tops = accumuloClient.tableOperations();
+ tops.create(table1);
+ tops.create(table2);
+
+ tops.offline(table1, true);
+ tops.offline(table2, true);
Review Comment:
Resolved in 83bc60a
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] cshannon commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1148421248
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -1285,4 +1304,153 @@ public BlockCacheConfiguration getBlockCacheConfiguration(AccumuloConfiguration
return BlockCacheConfiguration.forTabletServer(acuConf);
}
+ public int getOnDemandOnlineUnloadedForLowMemory() {
+ return onDemandUnloadedLowMemory.get();
+ }
+
+ // called from AssignmentHandler
+ public void insertOnDemandAccessTime(KeyExtent extent) {
+ onDemandTabletAccessTimes.putIfAbsent(extent, new AtomicLong(System.nanoTime()));
+ }
+
+ // called from getOnlineExtent
+ private void updateOnDemandAccessTime(KeyExtent extent) {
+ final long currentTime = System.nanoTime();
+ AtomicLong l = onDemandTabletAccessTimes.get(extent);
+ if (l != null) {
+ l.set(currentTime);
+ }
+ }
+
+ // called from UnloadTabletHandler
+ public void removeOnDemandAccessTime(KeyExtent extent) {
+ onDemandTabletAccessTimes.remove(extent);
+ }
+
+ private boolean isTabletInUse(KeyExtent extent) {
+ // Don't call getOnlineTablet as that will update the last access time
+ final Tablet t = onlineTablets.snapshot().get(extent);
+ if (t == null) {
+ return false;
+ }
+ t.updateRates(System.currentTimeMillis());
+ if (t.ingestRate() != 0.0 && t.queryRate() != 0.0 && t.scanRate() != 0.0) {
+ // tablet is ingesting or scanning
+ return true;
+ }
+ return false;
+ }
+
+ public void evaluateOnDemandTabletsForUnload() {
+
+ final SortedMap<KeyExtent,Tablet> online = getOnlineTablets();
+
+ // Find and remove access time entries for KeyExtents
+ // that are no longer in the onlineTablets collection
+ Set<KeyExtent> missing = onDemandTabletAccessTimes.keySet().stream()
+ .filter(k -> !online.containsKey(k)).collect(Collectors.toSet());
+ if (!missing.isEmpty()) {
+ log.debug("Removing onDemandAccessTimes for tablets as tablets no longer online: {}",
+ missing);
+ missing.forEach(onDemandTabletAccessTimes::remove);
+ if (onDemandTabletAccessTimes.isEmpty()) {
+ return;
+ }
+ }
+
+ // It's possible, from a tablet split or merge for example,
+ // that there is an onDemand tablet that is hosted for which
+ // we have no access time. Add any missing online onDemand
+ // tablets
+ online.forEach((k, v) -> {
+ if (v.isOnDemand() && !onDemandTabletAccessTimes.containsKey(k)) {
+ insertOnDemandAccessTime(k);
+ }
+ });
+
+ log.debug("Evaluating online onDemand tablets: {}", onDemandTabletAccessTimes);
+
+ if (onDemandTabletAccessTimes.isEmpty()) {
+ return;
+ }
+
+ // If the TabletServer is running low on memory, don't call the SPI
+ // plugin to evaluate which onDemand tablets to unload, just get the
+ // onDemand tablet with the oldest access time and unload it.
+ if (getContext().getLowMemoryDetector().isRunningLowOnMemory()) {
+ final SortedMap<Long,KeyExtent> timeSortedOnDemandExtents = new TreeMap<>();
+ onDemandTabletAccessTimes.forEach((k, v) -> timeSortedOnDemandExtents.put(v.get(), k));
+ Long oldestAccessTime = timeSortedOnDemandExtents.firstKey();
+ KeyExtent oldestKeyExtent = timeSortedOnDemandExtents.get(oldestAccessTime);
+ log.warn("Unloading onDemand tablet: {} for table: {} due to low memory", oldestKeyExtent,
+ oldestKeyExtent.tableId());
+ getContext().getAmple().mutateTablet(oldestKeyExtent).deleteOnDemand().mutate();
+ onDemandUnloadedLowMemory.addAndGet(1);
+ return;
+ }
+
+ // onDemandTabletAccessTimes is a HashMap. Sort the extents
+ // so that we can process them by table.
+ final SortedMap<KeyExtent,AtomicLong> sortedOnDemandExtents =
+ new TreeMap<KeyExtent,AtomicLong>();
+ sortedOnDemandExtents.putAll(onDemandTabletAccessTimes);
+
+ // The access times are updated when getOnlineTablet is called by other methods,
+ // but may not necessarily capture whether or not the Tablet is currently being used.
+ // For example, getOnlineTablet is called from startScan but not from continueScan.
+ // Instead of instrumenting all of the locations where the tablet is touched we
+ // can use the Tablet metrics.
+ final Set<KeyExtent> onDemandTabletsInUse = new HashSet<>();
+ for (KeyExtent extent : sortedOnDemandExtents.keySet()) {
+ if (isTabletInUse(extent)) {
+ onDemandTabletsInUse.add(extent);
+ }
+ }
+ if (!onDemandTabletsInUse.isEmpty()) {
+ log.debug("Removing onDemandAccessTimes for tablets as tablets are in use: {}",
+ onDemandTabletsInUse);
+ onDemandTabletsInUse.forEach(sortedOnDemandExtents::remove);
+ if (sortedOnDemandExtents.isEmpty()) {
+ return;
+ }
+ }
+
+ Set<TableId> tableIds = sortedOnDemandExtents.keySet().stream().map((k) -> {
+ return k.tableId();
+ }).distinct().collect(Collectors.toSet());
+ log.debug("Tables that have online onDemand tablets: {}", tableIds);
+ final Map<TableId,OnDemandTabletUnloader> unloaders = new HashMap<>();
+ tableIds.forEach(tid -> {
+ TableConfiguration tconf = getContext().getTableConfiguration(tid);
+ String tableContext = ClassLoaderUtil.tableContext(tconf);
+ String unloaderClassName = tconf.get(Property.TABLE_ONDEMAND_UNLOADER);
+ try {
+ Class<? extends OnDemandTabletUnloader> clazz = ClassLoaderUtil.loadClass(tableContext,
+ unloaderClassName, OnDemandTabletUnloader.class);
+ unloaders.put(tid, clazz.getConstructor().newInstance());
+ } catch (ClassNotFoundException | InstantiationException | IllegalAccessException
+ | IllegalArgumentException | InvocationTargetException | NoSuchMethodException
+ | SecurityException e) {
+ log.error(
+ "Error constructing OnDemandTabletUnloader implementation, not unloading onDemand tablets",
+ e);
+ return;
+ }
+ });
+ tableIds.forEach(tid -> {
+ Map<KeyExtent,AtomicLong> subset = sortedOnDemandExtents.entrySet().stream().filter((e) -> {
+ return e.getKey().tableId().equals(tid);
+ }).collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue()));
Review Comment:
```suggestion
AtomicLong> subset = sortedOnDemandExtents.entrySet().stream()
.filter((e) -> e.getKey().tableId().equals(tid))
.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
```
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java:
##########
@@ -1555,4 +1556,21 @@ public TSummaries contiuneGetSummaries(TInfo tinfo, long sessionId)
return handleTimeout(sessionId);
}
}
+
+ @Override
+ public void bringOnDemandTabletsOnline(TInfo tinfo, TCredentials credentials, String tableId,
+ List<TKeyExtent> extents) throws ThriftSecurityException, TException {
+ final TableId tid = TableId.of(tableId);
+ NamespaceId namespaceId = getNamespaceId(credentials, tid);
+ if (!security.canScan(credentials, tid, namespaceId)) {
+ throw new ThriftSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED);
+ }
+ try (TabletsMutator mutator = this.context.getAmple().mutateTablets()) {
+ extents.forEach(e -> {
+ mutator.mutateTablet(KeyExtent.fromThrift(e)).putOnDemand().mutate();
+ });
Review Comment:
```suggestion
extents.forEach(e -> mutator.mutateTablet(KeyExtent.fromThrift(e)).putOnDemand().mutate());
```
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocatorImpl.java:
##########
@@ -524,6 +541,72 @@ public TabletLocation locateTablet(ClientContext context, Text row, boolean skip
}
}
+ @Override
+ public long onDemandTabletsOnlined() {
+ return onDemandTabletsOnlinedCount.get();
+ }
+
+ private void bringOnDemandTabletsOnline(ClientContext context, Range range)
+ throws AccumuloException, AccumuloSecurityException {
+
+ // Confirm that table is in an onDemand state. Don't throw an exception
+ // if the table is not found, calling code will already handle it.
+ try {
+ String tableName = context.getTableName(tableId);
+ if (!context.tableOperations().isOnDemand(tableName)) {
+ log.trace("bringOnDemandTabletsOnline: table {} is not in ondemand state", tableId);
+ return;
+ }
+ } catch (TableNotFoundException e) {
+ log.trace("bringOnDemandTabletsOnline: table not found: {}", tableId);
+ return;
+ }
+
+ final Text scanRangeStart = range.getStartKey().getRow();
+ final Text scanRangeEnd = range.getEndKey().getRow();
+ // Turn the scan range into a KeyExtent and bring online all ondemand tablets
+ // that are overlapped by the scan range
+ final KeyExtent scanRangeKE = new KeyExtent(tableId, scanRangeEnd, scanRangeStart);
+
+ List<TKeyExtent> extentsToBringOnline = new ArrayList<>();
+
+ TabletsMetadata m = context.getAmple().readTablets().forTable(tableId)
+ .overlapping(scanRangeStart, true, null).build();
+ for (TabletMetadata tm : m) {
+ final KeyExtent tabletExtent = tm.getExtent();
+ log.trace("Evaluating tablet {} against range {}", tabletExtent, scanRangeKE);
+ if (tm.getEndRow() != null && tm.getEndRow().compareTo(scanRangeStart) < 0) {
+ // the end row of this tablet is before the start row, skip it
+ log.trace("tablet {} is before scan start range: {}", tabletExtent, scanRangeStart);
+ continue;
+ }
+ if (tm.getPrevEndRow() != null && tm.getPrevEndRow().compareTo(scanRangeEnd) > 0) {
+ // the start row of this tablet is after the scan range end row, skip it
+ log.trace("tablet {} is after scan end range: {}", tabletExtent, scanRangeEnd);
+ continue;
+ }
+ if (scanRangeKE.overlaps(tabletExtent)) {
+ if (!tm.getOnDemand()) {
+ Location loc = tm.getLocation();
+ if (loc != null) {
+ log.debug("tablet {} has location of: {}:{}", loc.getType(), loc.getHostPort());
Review Comment:
```suggestion
log.debug("tablet {} has location of: {}:{}", tabletExtent, loc.getType(),
loc.getHostPort());
```
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java:
##########
@@ -1555,4 +1556,21 @@ public TSummaries contiuneGetSummaries(TInfo tinfo, long sessionId)
return handleTimeout(sessionId);
}
}
+
+ @Override
+ public void bringOnDemandTabletsOnline(TInfo tinfo, TCredentials credentials, String tableId,
+ List<TKeyExtent> extents) throws ThriftSecurityException, TException {
Review Comment:
```suggestion
List<TKeyExtent> extents) throws TException {
```
##########
server/base/src/main/java/org/apache/accumulo/server/util/ListOnlineOnDemandTablets.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.util;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.TableNotFoundException;
+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.manager.state.tables.TableState;
+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.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.cli.ServerUtilOpts;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
+import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+public class ListOnlineOnDemandTablets {
+ private static final Logger log = LoggerFactory.getLogger(ListOnlineOnDemandTablets.class);
+
+ public static void main(String[] args) throws Exception {
+ ServerUtilOpts opts = new ServerUtilOpts();
+ opts.parseArgs(ListOnlineOnDemandTablets.class.getName(), args);
+ Span span = TraceUtil.startSpan(ListOnlineOnDemandTablets.class, "main");
+ try (Scope scope = span.makeCurrent()) {
+ ServerContext context = opts.getServerContext();
+ findOnline(context, null);
+ } finally {
+ span.end();
+ }
+ }
+
+ static int findOnline(ServerContext context, String tableName) throws TableNotFoundException {
+
+ final AtomicBoolean scanning = new AtomicBoolean(false);
+
+ LiveTServerSet tservers = new LiveTServerSet(context, new Listener() {
+ @Override
+ public void update(LiveTServerSet current, Set<TServerInstance> deleted,
+ Set<TServerInstance> added) {
+ if (!deleted.isEmpty() && scanning.get()) {
+ log.warn("Tablet servers deleted while scanning: {}", deleted);
+ }
+ if (!added.isEmpty() && scanning.get()) {
+ log.warn("Tablet servers added while scanning: {}", added);
+ }
+ }
+ });
+ tservers.startListeningForTabletServerChanges();
+ scanning.set(true);
+
+ System.out.println("Scanning " + MetadataTable.NAME);
+
+ Range range = TabletsSection.getRange();
+ if (tableName != null) {
+ TableId tableId = context.getTableId(tableName);
+ range = new KeyExtent(tableId, null, null).toMetaRange();
+ }
+
+ try (MetaDataTableScanner metaScanner =
+ new MetaDataTableScanner(context, range, MetadataTable.NAME)) {
+ return checkTablets(context, metaScanner, tservers);
+ }
+ }
+
+ private static int checkTablets(ServerContext context, Iterator<TabletLocationState> scanner,
+ LiveTServerSet tservers) {
+ int online = 0;
+
+ while (scanner.hasNext() && !System.out.checkError()) {
+ TabletLocationState locationState = scanner.next();
+ TabletState state = locationState.getState(tservers.getCurrentServers());
+ if (state != null && state == TabletState.HOSTED
+ && context.getTableManager().getTableState(locationState.extent.tableId())
Review Comment:
```suggestion
if (state == TabletState.HOSTED
&& context.getTableManager().getTableState(locationState.extent.tableId())
```
##########
server/base/src/main/java/org/apache/accumulo/server/util/ListOnlineOnDemandTablets.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.util;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.TableNotFoundException;
+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.manager.state.tables.TableState;
+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.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.cli.ServerUtilOpts;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
+import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+public class ListOnlineOnDemandTablets {
+ private static final Logger log = LoggerFactory.getLogger(ListOnlineOnDemandTablets.class);
+
+ public static void main(String[] args) throws Exception {
+ ServerUtilOpts opts = new ServerUtilOpts();
+ opts.parseArgs(ListOnlineOnDemandTablets.class.getName(), args);
+ Span span = TraceUtil.startSpan(ListOnlineOnDemandTablets.class, "main");
+ try (Scope scope = span.makeCurrent()) {
+ ServerContext context = opts.getServerContext();
+ findOnline(context, null);
Review Comment:
I assume the intent is to allow optionally passing in a table name as an argument? Right now this is always null and nowhere else is the method called.
##########
server/base/src/main/java/org/apache/accumulo/server/util/ListOnlineOnDemandTablets.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.util;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.TableNotFoundException;
+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.manager.state.tables.TableState;
+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.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.cli.ServerUtilOpts;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
+import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+public class ListOnlineOnDemandTablets {
+ private static final Logger log = LoggerFactory.getLogger(ListOnlineOnDemandTablets.class);
+
+ public static void main(String[] args) throws Exception {
+ ServerUtilOpts opts = new ServerUtilOpts();
+ opts.parseArgs(ListOnlineOnDemandTablets.class.getName(), args);
+ Span span = TraceUtil.startSpan(ListOnlineOnDemandTablets.class, "main");
+ try (Scope scope = span.makeCurrent()) {
+ ServerContext context = opts.getServerContext();
+ findOnline(context, null);
+ } finally {
+ span.end();
+ }
+ }
+
+ static int findOnline(ServerContext context, String tableName) throws TableNotFoundException {
+
+ final AtomicBoolean scanning = new AtomicBoolean(false);
+
+ LiveTServerSet tservers = new LiveTServerSet(context, new Listener() {
+ @Override
+ public void update(LiveTServerSet current, Set<TServerInstance> deleted,
+ Set<TServerInstance> added) {
+ if (!deleted.isEmpty() && scanning.get()) {
+ log.warn("Tablet servers deleted while scanning: {}", deleted);
+ }
+ if (!added.isEmpty() && scanning.get()) {
+ log.warn("Tablet servers added while scanning: {}", added);
+ }
+ }
+ });
+ tservers.startListeningForTabletServerChanges();
+ scanning.set(true);
+
+ System.out.println("Scanning " + MetadataTable.NAME);
+
+ Range range = TabletsSection.getRange();
+ if (tableName != null) {
+ TableId tableId = context.getTableId(tableName);
+ range = new KeyExtent(tableId, null, null).toMetaRange();
+ }
+
+ try (MetaDataTableScanner metaScanner =
+ new MetaDataTableScanner(context, range, MetadataTable.NAME)) {
+ return checkTablets(context, metaScanner, tservers);
+ }
+ }
+
+ private static int checkTablets(ServerContext context, Iterator<TabletLocationState> scanner,
+ LiveTServerSet tservers) {
+ int online = 0;
+
+ while (scanner.hasNext() && !System.out.checkError()) {
+ TabletLocationState locationState = scanner.next();
+ TabletState state = locationState.getState(tservers.getCurrentServers());
+ if (state != null && state == TabletState.HOSTED
+ && context.getTableManager().getTableState(locationState.extent.tableId())
Review Comment:
Small nit: Null check is redundant here
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -1285,4 +1304,153 @@ public BlockCacheConfiguration getBlockCacheConfiguration(AccumuloConfiguration
return BlockCacheConfiguration.forTabletServer(acuConf);
}
+ public int getOnDemandOnlineUnloadedForLowMemory() {
+ return onDemandUnloadedLowMemory.get();
+ }
+
+ // called from AssignmentHandler
+ public void insertOnDemandAccessTime(KeyExtent extent) {
+ onDemandTabletAccessTimes.putIfAbsent(extent, new AtomicLong(System.nanoTime()));
+ }
+
+ // called from getOnlineExtent
+ private void updateOnDemandAccessTime(KeyExtent extent) {
+ final long currentTime = System.nanoTime();
+ AtomicLong l = onDemandTabletAccessTimes.get(extent);
+ if (l != null) {
+ l.set(currentTime);
+ }
+ }
+
+ // called from UnloadTabletHandler
+ public void removeOnDemandAccessTime(KeyExtent extent) {
+ onDemandTabletAccessTimes.remove(extent);
+ }
+
+ private boolean isTabletInUse(KeyExtent extent) {
+ // Don't call getOnlineTablet as that will update the last access time
+ final Tablet t = onlineTablets.snapshot().get(extent);
+ if (t == null) {
+ return false;
+ }
+ t.updateRates(System.currentTimeMillis());
+ if (t.ingestRate() != 0.0 && t.queryRate() != 0.0 && t.scanRate() != 0.0) {
+ // tablet is ingesting or scanning
+ return true;
+ }
+ return false;
+ }
+
+ public void evaluateOnDemandTabletsForUnload() {
+
+ final SortedMap<KeyExtent,Tablet> online = getOnlineTablets();
+
+ // Find and remove access time entries for KeyExtents
+ // that are no longer in the onlineTablets collection
+ Set<KeyExtent> missing = onDemandTabletAccessTimes.keySet().stream()
+ .filter(k -> !online.containsKey(k)).collect(Collectors.toSet());
+ if (!missing.isEmpty()) {
+ log.debug("Removing onDemandAccessTimes for tablets as tablets no longer online: {}",
+ missing);
+ missing.forEach(onDemandTabletAccessTimes::remove);
+ if (onDemandTabletAccessTimes.isEmpty()) {
+ return;
+ }
+ }
+
+ // It's possible, from a tablet split or merge for example,
+ // that there is an onDemand tablet that is hosted for which
+ // we have no access time. Add any missing online onDemand
+ // tablets
+ online.forEach((k, v) -> {
+ if (v.isOnDemand() && !onDemandTabletAccessTimes.containsKey(k)) {
+ insertOnDemandAccessTime(k);
+ }
+ });
+
+ log.debug("Evaluating online onDemand tablets: {}", onDemandTabletAccessTimes);
+
+ if (onDemandTabletAccessTimes.isEmpty()) {
+ return;
+ }
+
+ // If the TabletServer is running low on memory, don't call the SPI
+ // plugin to evaluate which onDemand tablets to unload, just get the
+ // onDemand tablet with the oldest access time and unload it.
+ if (getContext().getLowMemoryDetector().isRunningLowOnMemory()) {
+ final SortedMap<Long,KeyExtent> timeSortedOnDemandExtents = new TreeMap<>();
+ onDemandTabletAccessTimes.forEach((k, v) -> timeSortedOnDemandExtents.put(v.get(), k));
+ Long oldestAccessTime = timeSortedOnDemandExtents.firstKey();
+ KeyExtent oldestKeyExtent = timeSortedOnDemandExtents.get(oldestAccessTime);
+ log.warn("Unloading onDemand tablet: {} for table: {} due to low memory", oldestKeyExtent,
+ oldestKeyExtent.tableId());
+ getContext().getAmple().mutateTablet(oldestKeyExtent).deleteOnDemand().mutate();
+ onDemandUnloadedLowMemory.addAndGet(1);
+ return;
+ }
+
+ // onDemandTabletAccessTimes is a HashMap. Sort the extents
+ // so that we can process them by table.
+ final SortedMap<KeyExtent,AtomicLong> sortedOnDemandExtents =
+ new TreeMap<KeyExtent,AtomicLong>();
+ sortedOnDemandExtents.putAll(onDemandTabletAccessTimes);
+
+ // The access times are updated when getOnlineTablet is called by other methods,
+ // but may not necessarily capture whether or not the Tablet is currently being used.
+ // For example, getOnlineTablet is called from startScan but not from continueScan.
+ // Instead of instrumenting all of the locations where the tablet is touched we
+ // can use the Tablet metrics.
+ final Set<KeyExtent> onDemandTabletsInUse = new HashSet<>();
+ for (KeyExtent extent : sortedOnDemandExtents.keySet()) {
+ if (isTabletInUse(extent)) {
+ onDemandTabletsInUse.add(extent);
+ }
+ }
+ if (!onDemandTabletsInUse.isEmpty()) {
+ log.debug("Removing onDemandAccessTimes for tablets as tablets are in use: {}",
+ onDemandTabletsInUse);
+ onDemandTabletsInUse.forEach(sortedOnDemandExtents::remove);
+ if (sortedOnDemandExtents.isEmpty()) {
+ return;
+ }
+ }
+
+ Set<TableId> tableIds = sortedOnDemandExtents.keySet().stream().map((k) -> {
+ return k.tableId();
+ }).distinct().collect(Collectors.toSet());
+ log.debug("Tables that have online onDemand tablets: {}", tableIds);
+ final Map<TableId,OnDemandTabletUnloader> unloaders = new HashMap<>();
+ tableIds.forEach(tid -> {
+ TableConfiguration tconf = getContext().getTableConfiguration(tid);
+ String tableContext = ClassLoaderUtil.tableContext(tconf);
+ String unloaderClassName = tconf.get(Property.TABLE_ONDEMAND_UNLOADER);
+ try {
+ Class<? extends OnDemandTabletUnloader> clazz = ClassLoaderUtil.loadClass(tableContext,
+ unloaderClassName, OnDemandTabletUnloader.class);
+ unloaders.put(tid, clazz.getConstructor().newInstance());
+ } catch (ClassNotFoundException | InstantiationException | IllegalAccessException
+ | IllegalArgumentException | InvocationTargetException | NoSuchMethodException
+ | SecurityException e) {
+ log.error(
+ "Error constructing OnDemandTabletUnloader implementation, not unloading onDemand tablets",
+ e);
+ return;
+ }
+ });
+ tableIds.forEach(tid -> {
+ Map<KeyExtent,AtomicLong> subset = sortedOnDemandExtents.entrySet().stream().filter((e) -> {
+ return e.getKey().tableId().equals(tid);
+ }).collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue()));
Review Comment:
This is just stylistic so don't need to change it if you don't want
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocatorImpl.java:
##########
@@ -524,6 +541,72 @@ public TabletLocation locateTablet(ClientContext context, Text row, boolean skip
}
}
+ @Override
+ public long onDemandTabletsOnlined() {
+ return onDemandTabletsOnlinedCount.get();
+ }
+
+ private void bringOnDemandTabletsOnline(ClientContext context, Range range)
+ throws AccumuloException, AccumuloSecurityException {
+
+ // Confirm that table is in an onDemand state. Don't throw an exception
+ // if the table is not found, calling code will already handle it.
+ try {
+ String tableName = context.getTableName(tableId);
+ if (!context.tableOperations().isOnDemand(tableName)) {
+ log.trace("bringOnDemandTabletsOnline: table {} is not in ondemand state", tableId);
+ return;
+ }
+ } catch (TableNotFoundException e) {
+ log.trace("bringOnDemandTabletsOnline: table not found: {}", tableId);
+ return;
+ }
+
+ final Text scanRangeStart = range.getStartKey().getRow();
+ final Text scanRangeEnd = range.getEndKey().getRow();
+ // Turn the scan range into a KeyExtent and bring online all ondemand tablets
+ // that are overlapped by the scan range
+ final KeyExtent scanRangeKE = new KeyExtent(tableId, scanRangeEnd, scanRangeStart);
+
+ List<TKeyExtent> extentsToBringOnline = new ArrayList<>();
+
+ TabletsMetadata m = context.getAmple().readTablets().forTable(tableId)
+ .overlapping(scanRangeStart, true, null).build();
+ for (TabletMetadata tm : m) {
+ final KeyExtent tabletExtent = tm.getExtent();
+ log.trace("Evaluating tablet {} against range {}", tabletExtent, scanRangeKE);
+ if (tm.getEndRow() != null && tm.getEndRow().compareTo(scanRangeStart) < 0) {
+ // the end row of this tablet is before the start row, skip it
+ log.trace("tablet {} is before scan start range: {}", tabletExtent, scanRangeStart);
+ continue;
+ }
+ if (tm.getPrevEndRow() != null && tm.getPrevEndRow().compareTo(scanRangeEnd) > 0) {
+ // the start row of this tablet is after the scan range end row, skip it
+ log.trace("tablet {} is after scan end range: {}", tabletExtent, scanRangeEnd);
+ continue;
+ }
+ if (scanRangeKE.overlaps(tabletExtent)) {
+ if (!tm.getOnDemand()) {
+ Location loc = tm.getLocation();
+ if (loc != null) {
+ log.debug("tablet {} has location of: {}:{}", loc.getType(), loc.getHostPort());
Review Comment:
There is a mismatched number of args here for the log statement, the tabletExtent is missing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147748253
##########
server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletStateChangeIterator.java:
##########
@@ -191,26 +194,29 @@ protected void consume() throws IOException {
}
// is the table supposed to be online or offline?
- boolean shouldBeOnline = onlineTables.contains(tls.extent.tableId());
+ final boolean shouldBeOnline = onlineTables.contains(tls.extent.tableId());
+ final boolean isOnDemandTable = onDemandTables.contains(tls.extent.tableId());
+ final boolean onDemandTabletShouldBeHosted = tls.ondemand;
Review Comment:
those variables are used in the debug log line that follows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147513513
##########
core/src/main/java/org/apache/accumulo/core/spi/ondemand/OnDemandTabletUnloader.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.spi.ondemand;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.TabletId;
+
+/**
+ * Object used by the TabletServer to determine which onDemand Tablets to unload for a Table
+ *
+ * @since 3.1.0
+ */
+public interface OnDemandTabletUnloader {
+
+ interface UnloaderParams {
+
+ /**
+ * @return table configuration
+ * @since 3.1.0
+ */
+ Map<String,String> getTableConfiguration();
+
+ /**
+ * Returns the onDemand tablets that are currently online and the time that they were last
+ * accessed
Review Comment:
Modified code and comments in 6409f84 to use System.nanoTime.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] cshannon commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1148424796
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java:
##########
@@ -1555,4 +1556,21 @@ public TSummaries contiuneGetSummaries(TInfo tinfo, long sessionId)
return handleTimeout(sessionId);
}
}
+
+ @Override
+ public void bringOnDemandTabletsOnline(TInfo tinfo, TCredentials credentials, String tableId,
+ List<TKeyExtent> extents) throws ThriftSecurityException, TException {
+ final TableId tid = TableId.of(tableId);
+ NamespaceId namespaceId = getNamespaceId(credentials, tid);
+ if (!security.canScan(credentials, tid, namespaceId)) {
+ throw new ThriftSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED);
+ }
+ try (TabletsMutator mutator = this.context.getAmple().mutateTablets()) {
+ extents.forEach(e -> {
+ mutator.mutateTablet(KeyExtent.fromThrift(e)).putOnDemand().mutate();
+ });
Review Comment:
Again, just a nit, feel free to ignore
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] cshannon commented on pull request #3250: Added On-Demand table feature
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#issuecomment-1483893531
I tested this out quite a bit today using Uno and so far so good. The instructions need to be updated though with new property names. I couldn't figure out why the tables were not unloading and then I did some debugging and it looks like things got renamed.
The instructions say to set:
```properties
manager.tablet.watcher.interval=15s
table.ondemand.tablet.unloader.interval=1m
table.custom.ondemand.unloader.inactivity.threshold=120000
```
But this should now be:
```properties
manager.tablet.watcher.interval=15s
#table renamed to tserver
tserver.ondemand.tablet.unloader.interval=1m
#seconds appended to the end
table.custom.ondemand.unloader.inactivity.threshold.seconds=120
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147830878
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,283 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ // Wait 2x the TabletGroupWatcher interval for ondemand
+ // tablets to be unassigned.
+ Thread.sleep(10000);
Review Comment:
Addressed in 8738077
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1145393278
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ BatchScanner s = c.createBatchScanner(tableName);
+ s.setRanges(Collections.singleton(scanRange));
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ BatchScanner s = c.createBatchScanner(tableName);
+ s.setRanges(Collections.singleton(scanRange));
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createBatchScanner(tableName);
+ s.setRanges(Collections.singleton(scanRange));
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
}
}
+ @Test
+ public void testBatchWriterAssignsTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
Resolved in 83bc60a
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ BatchScanner s = c.createBatchScanner(tableName);
+ s.setRanges(Collections.singleton(scanRange));
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
Resolved in 83bc60a
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion merged pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion merged PR #3250:
URL: https://github.com/apache/accumulo/pull/3250
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147441538
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -210,6 +215,8 @@ public PausedCompactionMetrics getPausedCompactionMetrics() {
private final AtomicLong syncCounter = new AtomicLong(0);
final OnlineTablets onlineTablets = new OnlineTablets();
+ private final Map<KeyExtent,AtomicLong> onDemandTabletAccessTimes =
Review Comment:
I have no issue with this, and is probably a better implementation. I wonder if this should be done as a follow-on commit because Tablet is going to change, maybe drastically. Thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1149187823
##########
server/base/src/main/java/org/apache/accumulo/server/util/ListOnlineOnDemandTablets.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.util;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.TableNotFoundException;
+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.manager.state.tables.TableState;
+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.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.cli.ServerUtilOpts;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
+import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+public class ListOnlineOnDemandTablets {
+ private static final Logger log = LoggerFactory.getLogger(ListOnlineOnDemandTablets.class);
+
+ public static void main(String[] args) throws Exception {
+ ServerUtilOpts opts = new ServerUtilOpts();
+ opts.parseArgs(ListOnlineOnDemandTablets.class.getName(), args);
+ Span span = TraceUtil.startSpan(ListOnlineOnDemandTablets.class, "main");
+ try (Scope scope = span.makeCurrent()) {
+ ServerContext context = opts.getServerContext();
+ findOnline(context, null);
Review Comment:
I didn't notice that. This class is a copy of FindOfflineTablets.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147570224
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -210,6 +215,8 @@ public PausedCompactionMetrics getPausedCompactionMetrics() {
private final AtomicLong syncCounter = new AtomicLong(0);
final OnlineTablets onlineTablets = new OnlineTablets();
+ private final Map<KeyExtent,AtomicLong> onDemandTabletAccessTimes =
Review Comment:
A follow on commit after merging SGTM. Could create an issue if needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147511781
##########
server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java:
##########
@@ -178,6 +188,13 @@ public synchronized void transitionTableState(final TableId tableId, final Table
log.error("FATAL Failed to transition table to state {}", newState);
throw new RuntimeException(e);
}
+ // Remove onDemand columns from all tablets
+ if (tableId != RootTable.ID && tableId != MetadataTable.ID
+ && (newState == TableState.ONLINE || newState == TableState.OFFLINE)) {
+ this.context.getAmple().readTablets().forTable(tableId).build().forEach(tm -> {
+ this.context.getAmple().mutateTablet(tm.getExtent()).deleteOnDemand().mutate();
+ });
Review Comment:
Updated in 6409f84
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1145338323
##########
test/src/main/java/org/apache/accumulo/test/MultiTableBatchWriterIT.java:
##########
@@ -400,4 +401,56 @@ public void testOfflineTable() throws Exception {
assertTrue(mutationsRejected, "Expected mutations to be rejected.");
}
+
+ @Test
+ public void testOnDemandTable() throws Exception {
+ boolean mutationsRejected = false;
+
+ try {
+ final String[] names = getUniqueNames(2);
+ final String table1 = names[0], table2 = names[1];
+
+ TableOperations tops = accumuloClient.tableOperations();
+ tops.create(table1);
+ tops.create(table2);
+
+ tops.offline(table1, true);
+ tops.offline(table2, true);
Review Comment:
I think these can be replaced with calls to `ondemand`. I believe that this test was created before the unload capability existed, so I used the `offline` command to unload the tablets.
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ BatchScanner s = c.createBatchScanner(tableName);
+ s.setRanges(Collections.singleton(scanRange));
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
I think these can be removed. I believe that this test was created before the unload capability existed, so I used the `offline` command to unload the tablets.
##########
test/src/main/java/org/apache/accumulo/test/functional/OnDemandIT.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.functional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.TabletLocator;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
+import org.apache.accumulo.test.metrics.TestStatsDSink;
+import org.apache.accumulo.test.metrics.TestStatsDSink.Metric;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class OnDemandIT extends SharedMiniClusterBase {
+
+ private static final int managerTabletGroupWatcherInterval = 5;
+ private static final int inactiveOnDemandTabletUnloaderInterval = 30;
+ private static TestStatsDSink sink;
+ private static Thread metricConsumer;
+ private static Long ONDEMAND_ONLINE_COUNT = 0L;
+
+ @BeforeAll
+ public static void beforeAll() throws Exception {
+ sink = new TestStatsDSink();
+ metricConsumer = new Thread(() -> {
+ while (!Thread.currentThread().isInterrupted()) {
+ List<String> statsDMetrics = sink.getLines();
+ for (String line : statsDMetrics) {
+ if (Thread.currentThread().isInterrupted()) {
+ break;
+ }
+ if (line.startsWith("accumulo")) {
+ Metric metric = TestStatsDSink.parseStatsDMetric(line);
+ if (MetricsProducer.METRICS_TSERVER_TABLETS_ONLINE_ONDEMAND.equals(metric.getName())) {
+ Long val = Long.parseLong(metric.getValue());
+ ONDEMAND_ONLINE_COUNT = val;
+ }
+ }
+ }
+ }
+ });
+ metricConsumer.start();
+ SharedMiniClusterBase.startMiniClusterWithConfig((cfg, core) -> {
+ cfg.setNumTservers(1);
+ cfg.setProperty(Property.MANAGER_TABLET_GROUP_WATCHER_INTERVAL,
+ Integer.toString(managerTabletGroupWatcherInterval));
+ cfg.setProperty(Property.TABLE_ONDEMAND_UNLOADER_INTERVAL,
+ Integer.toString(inactiveOnDemandTabletUnloaderInterval));
+ cfg.setProperty("table.custom.ondemand.unloader.inactivity.threshold", "30000");
+
+ // Tell the server processes to use a StatsDMeterRegistry that will be configured
+ // to push all metrics to the sink we started.
+ cfg.setProperty(Property.GENERAL_MICROMETER_ENABLED, "true");
+ cfg.setProperty(Property.GENERAL_MICROMETER_FACTORY,
+ TestStatsDRegistryFactory.class.getName());
+ Map<String,String> sysProps = Map.of(TestStatsDRegistryFactory.SERVER_HOST, "127.0.0.1",
+ TestStatsDRegistryFactory.SERVER_PORT, Integer.toString(sink.getPort()));
+ cfg.setSystemProperties(sysProps);
+ });
+ }
+
+ @AfterAll
+ public static void after() throws Exception {
+ sink.close();
+ metricConsumer.interrupt();
+ metricConsumer.join();
+ }
+
+ @BeforeEach
+ public void before() {
+ ONDEMAND_ONLINE_COUNT = 0L;
+ }
+
+ @Test
+ public void testLoadUnload() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ ManagerAssignmentIT.loadDataForScan(c, tableName);
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
I think these can be removed. I believe that this test was created before the unload capability existed, so I used the `offline` command to unload the tablets.
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
I think these can be removed. I believe that this test was created before the unload capability existed, so I used the `offline` command to unload the tablets.
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
I think these can be removed. I believe that this test was created before the unload capability existed, so I used the `offline` command to unload the tablets.
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
I think these can be removed. I believe that this test was created before the unload capability existed, so I used the `offline` command to unload the tablets.
##########
test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java:
##########
@@ -88,9 +119,278 @@ public void test() throws Exception {
assertNull(online.future);
assertNotNull(online.current);
assertEquals(online.current, online.last);
+
+ // take the tablet offline
+ c.tableOperations().offline(tableName, true);
+ offline = getTabletLocationState(c, tableId);
+ assertNull(offline.future);
+ assertNull(offline.current);
+ assertEquals(flushed.current, offline.last);
+
+ // set the table to ondemand
+ c.tableOperations().onDemand(tableName, true);
+ TabletLocationState ondemand = getTabletLocationState(c, tableId);
+ assertNull(ondemand.future);
+ assertNull(ondemand.current);
+ assertEquals(flushed.current, ondemand.last);
+
+ // put it back online
+ c.tableOperations().online(tableName, true);
+ online = getTabletLocationState(c, tableId);
+ assertNull(online.future);
+ assertNotNull(online.current);
+ assertEquals(online.current, online.last);
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ Scanner s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createScanner(tableName);
+ s.setRange(scanRange);
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsOneOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "c");
+ BatchScanner s = c.createBatchScanner(tableName);
+ s.setRanges(Collections.singleton(scanRange));
+ // Should return keys for a, b, c
+ assertEquals(3, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ // There should be one tablet online
+ assertEquals(1, stats.size());
+ assertEquals(1, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ }
+ }
+
+ @Test
+ public void testBatchScannerAssignsMultipleOnDemandTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
+ c.tableOperations().onDemand(tableName, true);
+ assertTrue(c.tableOperations().isOnDemand(tableName));
+
+ List<TabletStats> stats = getTabletStats(c, tableId);
+ // There should be no tablets online
+ assertEquals(0, stats.size());
+ assertEquals(0, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ c.tableOperations().clearLocatorCache(tableName);
+
+ Range scanRange = new Range("a", "s");
+ BatchScanner s = c.createBatchScanner(tableName);
+ s.setRanges(Collections.singleton(scanRange));
+ assertEquals(19, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
+ // Run another scan, all tablets should be loaded
+ scanRange = new Range("a", "t");
+ s = c.createBatchScanner(tableName);
+ s.setRanges(Collections.singleton(scanRange));
+ assertEquals(20, Iterables.size(s));
+
+ stats = getTabletStats(c, tableId);
+ assertEquals(3, stats.size());
+ // No more tablets should have been brought online
+ assertEquals(3, TabletLocator.getLocator((ClientContext) c, TableId.of(tableId))
+ .onDemandTabletsOnlined());
+
}
}
+ @Test
+ public void testBatchWriterAssignsTablets() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = super.getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ String tableId = c.tableOperations().tableIdMap().get(tableName);
+ loadDataForScan(c, tableName);
+
+ TreeSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("t"));
+ c.tableOperations().addSplits(tableName, splits);
+ // Need to offline the table first so that the tablets
+ // are unloaded.
+ c.tableOperations().offline(tableName, true);
Review Comment:
I think these can be removed. I believe that this test was created before the unload capability existed, so I used the `offline` command to unload the tablets.
##########
server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/LoadFiles.java:
##########
@@ -107,6 +107,7 @@ public long isReady(long tid, Manager manager) throws Exception {
@Override
public Repo<Manager> call(final long tid, final Manager manager) {
+ // TODO: How are we treating ONDEMAND tables for BulkImport?
Review Comment:
Need to address this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#issuecomment-1481053930
Ran a full IT build and a single test failed, ConditionalWriterIT.testDeleteTable. It failed because a TableNotFoundException was thrown instead of a TableDeletedException. This was due to a call being added to get `context.getTableName(TableId)` to check if the table was in an ondemand state. I moved this code in fa80774 and suppressed the exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1147708395
##########
core/src/main/java/org/apache/accumulo/core/conf/Property.java:
##########
@@ -309,6 +309,9 @@ public enum Property {
"The balancer class that accumulo will use to make tablet assignment and "
+ "migration decisions.",
"1.3.5"),
+ MANAGER_TABLET_GROUP_WATCHER_INTERVAL("manager.tablet.watcher.interval", "60s",
+ PropertyType.TIMEDURATION,
+ "Time to wait between scanning tablet states to determine migrations, etc.", "3.1.0"),
Review Comment:
Opened #3256 for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a diff in pull request #3250: Added On-Demand table feature
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1149207925
##########
server/base/src/main/java/org/apache/accumulo/server/util/ListOnlineOnDemandTablets.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.util;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.TableNotFoundException;
+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.manager.state.tables.TableState;
+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.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.cli.ServerUtilOpts;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
+import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+public class ListOnlineOnDemandTablets {
+ private static final Logger log = LoggerFactory.getLogger(ListOnlineOnDemandTablets.class);
+
+ public static void main(String[] args) throws Exception {
+ ServerUtilOpts opts = new ServerUtilOpts();
+ opts.parseArgs(ListOnlineOnDemandTablets.class.getName(), args);
+ Span span = TraceUtil.startSpan(ListOnlineOnDemandTablets.class, "main");
+ try (Scope scope = span.makeCurrent()) {
+ ServerContext context = opts.getServerContext();
+ findOnline(context, null);
Review Comment:
Addressed in 85d34f9
##########
server/base/src/main/java/org/apache/accumulo/server/util/ListOnlineOnDemandTablets.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.util;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.TableNotFoundException;
+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.manager.state.tables.TableState;
+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.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.cli.ServerUtilOpts;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
+import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+public class ListOnlineOnDemandTablets {
+ private static final Logger log = LoggerFactory.getLogger(ListOnlineOnDemandTablets.class);
+
+ public static void main(String[] args) throws Exception {
+ ServerUtilOpts opts = new ServerUtilOpts();
+ opts.parseArgs(ListOnlineOnDemandTablets.class.getName(), args);
+ Span span = TraceUtil.startSpan(ListOnlineOnDemandTablets.class, "main");
+ try (Scope scope = span.makeCurrent()) {
+ ServerContext context = opts.getServerContext();
+ findOnline(context, null);
+ } finally {
+ span.end();
+ }
+ }
+
+ static int findOnline(ServerContext context, String tableName) throws TableNotFoundException {
+
+ final AtomicBoolean scanning = new AtomicBoolean(false);
+
+ LiveTServerSet tservers = new LiveTServerSet(context, new Listener() {
+ @Override
+ public void update(LiveTServerSet current, Set<TServerInstance> deleted,
+ Set<TServerInstance> added) {
+ if (!deleted.isEmpty() && scanning.get()) {
+ log.warn("Tablet servers deleted while scanning: {}", deleted);
+ }
+ if (!added.isEmpty() && scanning.get()) {
+ log.warn("Tablet servers added while scanning: {}", added);
+ }
+ }
+ });
+ tservers.startListeningForTabletServerChanges();
+ scanning.set(true);
+
+ System.out.println("Scanning " + MetadataTable.NAME);
+
+ Range range = TabletsSection.getRange();
+ if (tableName != null) {
+ TableId tableId = context.getTableId(tableName);
+ range = new KeyExtent(tableId, null, null).toMetaRange();
+ }
+
+ try (MetaDataTableScanner metaScanner =
+ new MetaDataTableScanner(context, range, MetadataTable.NAME)) {
+ return checkTablets(context, metaScanner, tservers);
+ }
+ }
+
+ private static int checkTablets(ServerContext context, Iterator<TabletLocationState> scanner,
+ LiveTServerSet tservers) {
+ int online = 0;
+
+ while (scanner.hasNext() && !System.out.checkError()) {
+ TabletLocationState locationState = scanner.next();
+ TabletState state = locationState.getState(tservers.getCurrentServers());
+ if (state != null && state == TabletState.HOSTED
+ && context.getTableManager().getTableState(locationState.extent.tableId())
Review Comment:
Addressed in 85d34f9
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org