You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by GitBox <gi...@apache.org> on 2022/11/21 20:00:26 UTC
[GitHub] [accumulo] dlmarion opened a new pull request, #3091: Added code and tests to confirm ScanServers can scan offline tables
dlmarion opened a new pull request, #3091:
URL: https://github.com/apache/accumulo/pull/3091
The TabletLocator only caches locations for tablets that are online. Had to add code to find the KeyExtent in the metadata table for the scan range when the tablets were not in the TabletLocator. Modified code paths to throw TableOfflineException when scanning using Tablet Servers only.
Closes #3084
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1329692249
> The purpose of this test is to ensure that the client side code does not keep caching extents that no longer exist.
Regarding this - how would the client know whether the cache contents are correct or not if it doesn't query the metadata table each time to compare the cached data?
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1033628645
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchReaderIterator.java:
##########
@@ -241,38 +247,59 @@ private void binRanges(TabletLocator tabletLocator, List<Range> ranges,
int lastFailureSize = Integer.MAX_VALUE;
- while (true) {
+ if (options.getConsistencyLevel().equals(ConsistencyLevel.EVENTUAL) && !this.tableOnline) {
+ // The TabletLocator only caches the location of tablets for online tables. The ScanServer
+ // can scan online and offline tables, but does not require the location of the tablet. It
+ // only requires the KeyExtent. Find the first extent for the table that matches the
+ // startRow provided in the scanState. If no startRow is provided, use the first KeyExtent
+ // for the table
binnedRanges.clear();
- List<Range> failures = tabletLocator.binRanges(context, ranges, binnedRanges);
-
- if (failures.isEmpty()) {
- break;
- } else {
- // tried to only do table state checks when failures.size() == ranges.size(), however this
- // did
- // not work because nothing ever invalidated entries in the tabletLocator cache... so even
- // though
- // the table was deleted the tablet locator entries for the deleted table were not
- // cleared... so
- // need to always do the check when failures occur
- if (failures.size() >= lastFailureSize) {
- context.requireNotDeleted(tableId);
- context.requireNotOffline(tableId, tableName);
+ Map<KeyExtent,List<Range>> map = new HashMap<>();
+ ranges.forEach(r -> {
+ TabletsMetadata tm = context.getAmple().readTablets().forTable(this.tableId)
+ .overlapping(r.getStartKey() == null ? null : r.getStartKey().getRow(),
+ r.isStartKeyInclusive(), r.getEndKey() == null ? null : r.getEndKey().getRow())
+ .build();
Review Comment:
I found the KeyExtentCache in BulkImport. It looks like I should be able to reuse this, although I will likely have to move it to a different package. I may also need to modify it as it looks like the cached extents do not expire.
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1339239073
Full IT build succeeded
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1033939311
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchReaderIterator.java:
##########
@@ -241,38 +247,59 @@ private void binRanges(TabletLocator tabletLocator, List<Range> ranges,
int lastFailureSize = Integer.MAX_VALUE;
- while (true) {
+ if (options.getConsistencyLevel().equals(ConsistencyLevel.EVENTUAL) && !this.tableOnline) {
+ // The TabletLocator only caches the location of tablets for online tables. The ScanServer
+ // can scan online and offline tables, but does not require the location of the tablet. It
+ // only requires the KeyExtent. Find the first extent for the table that matches the
+ // startRow provided in the scanState. If no startRow is provided, use the first KeyExtent
+ // for the table
binnedRanges.clear();
- List<Range> failures = tabletLocator.binRanges(context, ranges, binnedRanges);
-
- if (failures.isEmpty()) {
- break;
- } else {
- // tried to only do table state checks when failures.size() == ranges.size(), however this
- // did
- // not work because nothing ever invalidated entries in the tabletLocator cache... so even
- // though
- // the table was deleted the tablet locator entries for the deleted table were not
- // cleared... so
- // need to always do the check when failures occur
- if (failures.size() >= lastFailureSize) {
- context.requireNotDeleted(tableId);
- context.requireNotOffline(tableId, tableName);
+ Map<KeyExtent,List<Range>> map = new HashMap<>();
+ ranges.forEach(r -> {
+ TabletsMetadata tm = context.getAmple().readTablets().forTable(this.tableId)
+ .overlapping(r.getStartKey() == null ? null : r.getStartKey().getRow(),
+ r.isStartKeyInclusive(), r.getEndKey() == null ? null : r.getEndKey().getRow())
+ .build();
Review Comment:
@keith-turner - I created a new, different cache class in d73b879 that has different logic and implementation than the ConcurrentKeyExtentCache in the bulk import code. I'm going to kick off an IT run to ensure that this cache is working (I think it is based on the limited IT tests that I ran locally). Would be interested in your feedback on the logic in the new cache and whether or not it can replace the cache in the bulk import code.
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1335171490
@ctubbsii - That's correct. I reverted the prior change from 2.1 (and main by merge of 2.1 -> main). I figured I would make the changes here so that an incomplete and non-functional change was not included in a 2.1 patch release.
--
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] ctubbsii commented on pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
ctubbsii commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1324539493
@dlmarion Is this targeting 3.0? It's based on the main branch instead of 2.1 branch. The issue this closes says it's planned for 2.1.1.
--
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] ctubbsii commented on a diff in pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
ctubbsii commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1035123963
##########
core/src/main/thrift/tabletserver.thrift:
##########
@@ -265,6 +265,8 @@ service TabletScanClientService {
1:client.ThriftSecurityException sec
2:TSampleNotPresentException tsnpe
3:ScanServerBusyException ssbe
+ 4:NotServingTabletException nste
+ 5:TooManyFilesException tmfe
Review Comment:
One of the expectations in a patch release is that we can do rolling upgrades, and clients and servers may be upgraded at different times because patch releases are both forwards and backwards compatible. These are both already TExceptions, which should be caught in the generated code as a generic catch-all. However, I'm not sure what the exact behavior would be when these are declared explicitly in 2.1.1, but not in 2.1.0 if a client and server of different versions are talking to each other.
Before approving this, I would like to know that the behavior is sensible if there is a 2.1.0 client talking to a 2.1.1 server, or a 2.1.1 client talking to a 2.1.0 server when these exceptions are thrown. Have you tested these scenarios? What is the impact?
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1042107655
##########
core/src/main/java/org/apache/accumulo/core/data/RangedObject.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.data;
+
+/**
+ * Marker interface that declares the object is a type that represents a range
+ */
+public interface RangedObject<T> {
Review Comment:
I have no issue removing the interface. I added it because @keith-turner had mentioned to me that he was thinking of creating a generic TreeMap-based cache that checks for overlaps.
--
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] ctubbsii commented on pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
ctubbsii commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1334580135
@dlmarion I think you're planning on re-applying the changes that were reverted in #3101 to this PR. Is that correct?
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1028472028
##########
test/src/main/java/org/apache/accumulo/test/ScanServerNoSServersIT.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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;
+
+import static org.apache.accumulo.harness.AccumuloITBase.MINI_CLUSTER_ONLY;
+import static org.apache.accumulo.test.ScanServerIT.createTableAndIngest;
+import static org.apache.accumulo.test.ScanServerIT.ingest;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Collections;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.ScannerBase.ConsistencyLevel;
+import org.apache.accumulo.core.client.TableOfflineException;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+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.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import com.google.common.collect.Iterables;
+
+@Tag(MINI_CLUSTER_ONLY)
+public class ScanServerNoSServersIT extends SharedMiniClusterBase {
+
+ // This is the same as ScanServerIT, but without any scan servers running.
+ // This tests the cases where the client falls back to the Tablet Servers
+
+ private static class ScanServerITConfiguration implements MiniClusterConfigurationCallback {
+
+ @Override
+ public void configureMiniCluster(MiniAccumuloConfigImpl cfg,
+ org.apache.hadoop.conf.Configuration coreSite) {
+ cfg.setNumScanServers(0);
+
+ // Timeout scan sessions after being idle for 3 seconds
+ cfg.setProperty(Property.TSERV_SESSION_MAXIDLE, "3s");
+
+ // Configure the scan server to only have 1 scan executor thread. This means
+ // that the scan server will run scans serially, not concurrently.
+ cfg.setProperty(Property.SSERV_SCAN_EXECUTORS_DEFAULT_THREADS, "1");
+ }
+ }
+
+ @BeforeAll
+ public static void start() throws Exception {
+ ScanServerITConfiguration c = new ScanServerITConfiguration();
+ SharedMiniClusterBase.startMiniClusterWithConfig(c);
+ }
+
+ @AfterAll
+ public static void stop() throws Exception {
+ SharedMiniClusterBase.stopMiniCluster();
+ }
+
+ @Test
+ public void testScan() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", false);
+
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "Scanner did not see ingested and flushed entries");
+ final int additionalIngestedEntryCount =
+ ingest(client, tableName, 10, 10, 10, "colf", false, 1);
+ // since there are no scan servers, and we are reading from tservers, we should see update
+ assertEquals(ingestedEntryCount + additionalIngestedEntryCount, Iterables.size(scanner),
+ "Scanning against tserver should have resulted in seeing all ingested entries");
+ } // when the scanner is closed, all open sessions should be closed
+ }
+ }
+
+ @Test
+ public void testScanOfflineSingleTablet() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", false);
+ client.tableOperations().offline(tableName, true);
+ assertFalse(client.tableOperations().isOnline(tableName));
+
+ TabletLocator.getLocator((ClientContext) client,
+ TableId.of(client.tableOperations().tableIdMap().get(tableName))).invalidateCache();
+
+ assertThrows(TableOfflineException.class, () -> {
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "The scan server scanner should have seen all ingested and flushed entries");
+ } // when the scanner is closed, all open sessions should be closed
+ });
+ }
+ }
+
+ @Test
+ public void testScanOfflineTableManyTablets() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", true);
+ client.tableOperations().offline(tableName, true);
+ assertFalse(client.tableOperations().isOnline(tableName));
+
+ TabletLocator.getLocator((ClientContext) client,
+ TableId.of(client.tableOperations().tableIdMap().get(tableName))).invalidateCache();
+
+ assertThrows(TableOfflineException.class, () -> {
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "The scan server scanner should have seen all ingested and flushed entries");
+ } // when the scanner is closed, all open sessions should be closed
+ });
+ }
+ }
+
+ @Test
+ public void testScanTableCreatedOffline() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+ client.tableOperations().create(tableName, new NewTableConfiguration().createOffline());
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
Review Comment:
This doesn't throw a TableOfflineException, need to confirm that this is consistent with current behavior
##########
test/src/main/java/org/apache/accumulo/test/ScanServerNoSServersIT.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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;
+
+import static org.apache.accumulo.harness.AccumuloITBase.MINI_CLUSTER_ONLY;
+import static org.apache.accumulo.test.ScanServerIT.createTableAndIngest;
+import static org.apache.accumulo.test.ScanServerIT.ingest;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Collections;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.ScannerBase.ConsistencyLevel;
+import org.apache.accumulo.core.client.TableOfflineException;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+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.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import com.google.common.collect.Iterables;
+
+@Tag(MINI_CLUSTER_ONLY)
+public class ScanServerNoSServersIT extends SharedMiniClusterBase {
+
+ // This is the same as ScanServerIT, but without any scan servers running.
+ // This tests the cases where the client falls back to the Tablet Servers
+
+ private static class ScanServerITConfiguration implements MiniClusterConfigurationCallback {
+
+ @Override
+ public void configureMiniCluster(MiniAccumuloConfigImpl cfg,
+ org.apache.hadoop.conf.Configuration coreSite) {
+ cfg.setNumScanServers(0);
+
+ // Timeout scan sessions after being idle for 3 seconds
+ cfg.setProperty(Property.TSERV_SESSION_MAXIDLE, "3s");
+
+ // Configure the scan server to only have 1 scan executor thread. This means
+ // that the scan server will run scans serially, not concurrently.
+ cfg.setProperty(Property.SSERV_SCAN_EXECUTORS_DEFAULT_THREADS, "1");
+ }
+ }
+
+ @BeforeAll
+ public static void start() throws Exception {
+ ScanServerITConfiguration c = new ScanServerITConfiguration();
+ SharedMiniClusterBase.startMiniClusterWithConfig(c);
+ }
+
+ @AfterAll
+ public static void stop() throws Exception {
+ SharedMiniClusterBase.stopMiniCluster();
+ }
+
+ @Test
+ public void testScan() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", false);
+
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "Scanner did not see ingested and flushed entries");
+ final int additionalIngestedEntryCount =
+ ingest(client, tableName, 10, 10, 10, "colf", false, 1);
+ // since there are no scan servers, and we are reading from tservers, we should see update
+ assertEquals(ingestedEntryCount + additionalIngestedEntryCount, Iterables.size(scanner),
+ "Scanning against tserver should have resulted in seeing all ingested entries");
+ } // when the scanner is closed, all open sessions should be closed
+ }
+ }
+
+ @Test
+ public void testScanOfflineSingleTablet() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", false);
+ client.tableOperations().offline(tableName, true);
+ assertFalse(client.tableOperations().isOnline(tableName));
+
+ TabletLocator.getLocator((ClientContext) client,
+ TableId.of(client.tableOperations().tableIdMap().get(tableName))).invalidateCache();
+
+ assertThrows(TableOfflineException.class, () -> {
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "The scan server scanner should have seen all ingested and flushed entries");
+ } // when the scanner is closed, all open sessions should be closed
+ });
+ }
+ }
+
+ @Test
+ public void testScanOfflineTableManyTablets() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", true);
+ client.tableOperations().offline(tableName, true);
+ assertFalse(client.tableOperations().isOnline(tableName));
+
+ TabletLocator.getLocator((ClientContext) client,
+ TableId.of(client.tableOperations().tableIdMap().get(tableName))).invalidateCache();
+
+ assertThrows(TableOfflineException.class, () -> {
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "The scan server scanner should have seen all ingested and flushed entries");
+ } // when the scanner is closed, all open sessions should be closed
+ });
+ }
+ }
+
+ @Test
+ public void testScanTableCreatedOffline() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+ client.tableOperations().create(tableName, new NewTableConfiguration().createOffline());
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ } // when the scanner is closed, all open sessions should be closed
+ }
+ }
+
+ @Test
+ public void testBatchScan() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", false);
+
+ try (BatchScanner scanner = client.createBatchScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRanges(Collections.singletonList(new Range()));
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "Scanner did not see ingested and flushed entries");
+ final int additionalIngestedEntryCount =
+ ingest(client, tableName, 10, 10, 10, "colf", false, 1);
+ // since there are no scan servers, and we are reading from tservers, we should see update
+ assertEquals(ingestedEntryCount + additionalIngestedEntryCount, Iterables.size(scanner),
+ "Scanning against tserver should have resulted in seeing all ingested entries");
+ } // when the scanner is closed, all open sessions should be closed
+ }
+ }
+
+ @Test
+ public void testBatchScanTableCreatedOffline() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+ client.tableOperations().create(tableName, new NewTableConfiguration().createOffline());
+
+ try (BatchScanner scanner = client.createBatchScanner(tableName, Authorizations.EMPTY)) {
Review Comment:
This doesn't throw a TableOfflineException, need to confirm that this is consistent with current behavior
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1324214044
Full IT build passed
--
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 closed pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion closed pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
URL: https://github.com/apache/accumulo/pull/3091
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1044729362
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/ScannerImpl.java:
##########
@@ -152,6 +153,16 @@ public synchronized int getBatchSize() {
@Override
public synchronized Iterator<Entry<Key,Value>> iterator() {
ensureOpen();
+
+ if (getConsistencyLevel() == ConsistencyLevel.IMMEDIATE) {
+ try {
+ String tableName = context.getTableName(tableId);
+ context.requireNotOffline(tableId, tableName);
+ } catch (TableNotFoundException e) {
+ throw new RuntimeException("Table not found", e);
Review Comment:
Created UncheckedAccumuloException in 5b98e2a
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftScanner.java:
##########
@@ -559,6 +580,11 @@ public Map<String,String> getHints() {
"For tablet {} scan server selector chose scan_server:{} delay:{} busyTimeout:{}",
loc.tablet_extent, scanServer, delay, scanState.busyTimeout);
} else {
+ try {
+ context.requireNotOffline(scanState.tableId, context.getTableName(scanState.tableId));
+ } catch (TableNotFoundException e) {
+ throw new RuntimeException("Error scanning for table: " + scanState.tableId, e);
Review Comment:
Created UncheckedAccumuloException in 5b98e2a
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1324936452
@ctubbsii - Thanks, I didn't notice that. I changed the base branch.
--
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] ctubbsii commented on a diff in pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
ctubbsii commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1042982025
##########
core/src/main/java/org/apache/accumulo/core/data/RangedObject.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.data;
+
+/**
+ * Marker interface that declares the object is a type that represents a range
+ */
+public interface RangedObject<T> {
Review Comment:
Let's wait to include it alongside that proposed change, if/when it manifests. I think it's premature here, as its API needs might be slightly different than what you have here by the time it is ready.
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1044729050
##########
core/src/main/java/org/apache/accumulo/core/data/RangedObject.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.data;
+
+/**
+ * Marker interface that declares the object is a type that represents a range
+ */
+public interface RangedObject<T> {
Review Comment:
Removed RangedObject in 5b98e2a
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
keith-turner commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1042082740
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/OfflineTableExtentCache.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.clientImpl;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeMap;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.hadoop.io.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Scheduler;
+
+public class OfflineTableExtentCache {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OfflineTableExtentCache.class);
+
+ private final Cache<TableId,TreeMap<Range,List<KeyExtent>>> cache;
+ private final ClientContext ctx;
+
+ OfflineTableExtentCache(ClientContext ctx) {
+ this.ctx = ctx;
+ cache = Caffeine.newBuilder().scheduler(Scheduler.systemScheduler())
+ .expireAfterAccess(Duration.ofMinutes(10)).build();
+ }
+
+ private List<KeyExtent> loadExtentsForRange(TableId tid, Range r) {
+ LOG.trace("Loading extents for tid: {} and range: {}", tid, r);
+ TabletsMetadata tm =
+ ctx.getAmple().readTablets().forTable(tid)
+ .overlapping(r.getStartKey() == null ? null : r.getStartKey().getRow(),
+ r.isStartKeyInclusive(), r.getEndKey() == null ? null : r.getEndKey().getRow())
+ .build();
+ List<KeyExtent> result = new ArrayList<>();
+ for (TabletMetadata t : tm) {
+ LOG.trace("Evaluating: {}", t.getExtent());
+ if (t.getExtent().endRow() == null || r.getEndKey() == null || t.getExtent().endRow() != null
+ && r.getEndKey().compareTo(new Key(t.getExtent().endRow()), PartialKey.ROW) >= 0) {
+ LOG.trace("Adding: {}", t.getExtent());
+ result.add(t.getExtent());
+ }
+ }
+ LOG.trace("Loaded extents: {}", result);
+ return result;
+ }
+
+ public Map<Range,List<KeyExtent>> lookup(TableId tid, List<Range> ranges) {
Review Comment:
Not completely sure, but I think this method could result in returning overlapping key extents for a range and/or missing extents.
For overlapping tablets, could the following happen?
* Range R1 is added to cache with extent E1.
* Range R2 is added to the cache with extents E1b. Range R2 does not overlap with R1. E1b is the result of E1 splitting.
* A lookup is done for range R3 that overlaps R1 and R2. Will it return extents E1 and E1b for R3?
For missing a tablet could the following happen?
* Assume a table has four contiguous tablets : E2,E3,E4,E5
* A lookup for range R3 that only overlaps E2 adds R3->[E2] to the cache.
* A lookup for range R4 that only overlaps E4 adds R4->[E4] to the cache.
* A lookup for range R5 that only overlaps E5 adds R5->[E5] to the tablet.
* A lookup for range R6 arrives. R6 overlaps R3,R4,and R5 and it also overlaps E2,E3,E4,and E5. Would the cache lookup only return E2,E4,E5 for R6?
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/OfflineTableExtentCache.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.clientImpl;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeMap;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.hadoop.io.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Scheduler;
+
+public class OfflineTableExtentCache {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OfflineTableExtentCache.class);
+
+ private final Cache<TableId,TreeMap<Range,List<KeyExtent>>> cache;
+ private final ClientContext ctx;
+
+ OfflineTableExtentCache(ClientContext ctx) {
+ this.ctx = ctx;
+ cache = Caffeine.newBuilder().scheduler(Scheduler.systemScheduler())
+ .expireAfterAccess(Duration.ofMinutes(10)).build();
+ }
+
+ private List<KeyExtent> loadExtentsForRange(TableId tid, Range r) {
+ LOG.trace("Loading extents for tid: {} and range: {}", tid, r);
+ TabletsMetadata tm =
+ ctx.getAmple().readTablets().forTable(tid)
+ .overlapping(r.getStartKey() == null ? null : r.getStartKey().getRow(),
+ r.isStartKeyInclusive(), r.getEndKey() == null ? null : r.getEndKey().getRow())
+ .build();
+ List<KeyExtent> result = new ArrayList<>();
+ for (TabletMetadata t : tm) {
+ LOG.trace("Evaluating: {}", t.getExtent());
+ if (t.getExtent().endRow() == null || r.getEndKey() == null || t.getExtent().endRow() != null
+ && r.getEndKey().compareTo(new Key(t.getExtent().endRow()), PartialKey.ROW) >= 0) {
+ LOG.trace("Adding: {}", t.getExtent());
+ result.add(t.getExtent());
+ }
+ }
+ LOG.trace("Loaded extents: {}", result);
+ return result;
+ }
+
+ public Map<Range,List<KeyExtent>> lookup(TableId tid, List<Range> ranges) {
+ LOG.trace("Performing lookup for table: {}, range: {}", tid, ranges);
+ Map<Range,List<KeyExtent>> results = new TreeMap<>();
+ TreeMap<Range,List<KeyExtent>> extentMap = cache.get(tid, (t) -> {
+ return new TreeMap<Range,List<KeyExtent>>();
+ });
+
+ Collections.sort(ranges);
+ ranges.forEach(r -> {
+ // Find an existing range in the extentMap that overlaps the scan range
+ boolean foundRange = false;
+ Range rangeBefore = extentMap.floorKey(r);
+ if (rangeBefore != null) {
+ Map<Range,List<KeyExtent>> tm = extentMap.tailMap(rangeBefore);
+ for (Entry<Range,List<KeyExtent>> e : tm.entrySet()) {
+ if (e.getKey().overlaps(r)) {
+ // Ignore extents that don't contain the scan range
+ List<KeyExtent> extents = new ArrayList<>();
+ if (r.getStartKey() == null) {
+ extents.addAll(e.getValue());
+ } else {
+ e.getValue().forEach(extent -> {
+ Text row = r.isStartKeyInclusive() ? r.getStartKey().getRow()
+ : r.getStartKey().followingKey(PartialKey.ROW).getRow();
+ if (extent.contains(row)) {
+ extents.add(extent);
+ }
+ });
+ }
+ results.put(r, extents);
+ foundRange = true;
+ break;
+ }
+ }
+ }
+ if (!foundRange) {
+ List<KeyExtent> extents = loadExtentsForRange(tid, r);
+ if (extents != null) {
+ extentMap.put(r, extents);
+ results.put(r, extents);
+ }
+ }
+ });
+ // remove any overlapping from the cache
+ removeOverlapping(extentMap);
+ LOG.trace("Returning extents: {}", extentMap);
+ return results;
+ }
+
+ private void removeOverlapping(TreeMap<Range,List<KeyExtent>> map) {
+
+ Range idx = map.firstKey();
+ Set<Range> removals = new HashSet<>();
+ for (Range range : map.keySet()) {
Review Comment:
Seems like this is an O(N) operation that will occur on each cache lookup.
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/OfflineTableExtentCache.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.clientImpl;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeMap;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.hadoop.io.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Scheduler;
+
+public class OfflineTableExtentCache {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OfflineTableExtentCache.class);
+
+ private final Cache<TableId,TreeMap<Range,List<KeyExtent>>> cache;
Review Comment:
There could be concurrency issues if multiple threads are accessing the treemap in the cache.
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/OfflineTableExtentCache.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.clientImpl;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeMap;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.hadoop.io.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Scheduler;
+
+public class OfflineTableExtentCache {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OfflineTableExtentCache.class);
+
+ private final Cache<TableId,TreeMap<Range,List<KeyExtent>>> cache;
+ private final ClientContext ctx;
+
+ OfflineTableExtentCache(ClientContext ctx) {
+ this.ctx = ctx;
+ cache = Caffeine.newBuilder().scheduler(Scheduler.systemScheduler())
+ .expireAfterAccess(Duration.ofMinutes(10)).build();
+ }
+
+ private List<KeyExtent> loadExtentsForRange(TableId tid, Range r) {
+ LOG.trace("Loading extents for tid: {} and range: {}", tid, r);
+ TabletsMetadata tm =
+ ctx.getAmple().readTablets().forTable(tid)
+ .overlapping(r.getStartKey() == null ? null : r.getStartKey().getRow(),
+ r.isStartKeyInclusive(), r.getEndKey() == null ? null : r.getEndKey().getRow())
+ .build();
+ List<KeyExtent> result = new ArrayList<>();
+ for (TabletMetadata t : tm) {
+ LOG.trace("Evaluating: {}", t.getExtent());
+ if (t.getExtent().endRow() == null || r.getEndKey() == null || t.getExtent().endRow() != null
+ && r.getEndKey().compareTo(new Key(t.getExtent().endRow()), PartialKey.ROW) >= 0) {
+ LOG.trace("Adding: {}", t.getExtent());
+ result.add(t.getExtent());
+ }
+ }
+ LOG.trace("Loaded extents: {}", result);
+ return result;
+ }
+
+ public Map<Range,List<KeyExtent>> lookup(TableId tid, List<Range> ranges) {
+ LOG.trace("Performing lookup for table: {}, range: {}", tid, ranges);
+ Map<Range,List<KeyExtent>> results = new TreeMap<>();
+ TreeMap<Range,List<KeyExtent>> extentMap = cache.get(tid, (t) -> {
+ return new TreeMap<Range,List<KeyExtent>>();
+ });
+
+ Collections.sort(ranges);
+ ranges.forEach(r -> {
+ // Find an existing range in the extentMap that overlaps the scan range
+ boolean foundRange = false;
+ Range rangeBefore = extentMap.floorKey(r);
+ if (rangeBefore != null) {
+ Map<Range,List<KeyExtent>> tm = extentMap.tailMap(rangeBefore);
+ for (Entry<Range,List<KeyExtent>> e : tm.entrySet()) {
+ if (e.getKey().overlaps(r)) {
+ // Ignore extents that don't contain the scan range
+ List<KeyExtent> extents = new ArrayList<>();
+ if (r.getStartKey() == null) {
+ extents.addAll(e.getValue());
+ } else {
+ e.getValue().forEach(extent -> {
+ Text row = r.isStartKeyInclusive() ? r.getStartKey().getRow()
+ : r.getStartKey().followingKey(PartialKey.ROW).getRow();
+ if (extent.contains(row)) {
Review Comment:
Will this select extents that fall completely within the query range? Like a query range that is [b,r] and and extent that is (j,m]? The extent (j,m] will not contain the start row b.
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/OfflineTableExtentCache.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.clientImpl;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeMap;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.hadoop.io.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Scheduler;
+
+public class OfflineTableExtentCache {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OfflineTableExtentCache.class);
+
+ private final Cache<TableId,TreeMap<Range,List<KeyExtent>>> cache;
+ private final ClientContext ctx;
+
+ OfflineTableExtentCache(ClientContext ctx) {
+ this.ctx = ctx;
+ cache = Caffeine.newBuilder().scheduler(Scheduler.systemScheduler())
+ .expireAfterAccess(Duration.ofMinutes(10)).build();
+ }
+
+ private List<KeyExtent> loadExtentsForRange(TableId tid, Range r) {
+ LOG.trace("Loading extents for tid: {} and range: {}", tid, r);
+ TabletsMetadata tm =
+ ctx.getAmple().readTablets().forTable(tid)
+ .overlapping(r.getStartKey() == null ? null : r.getStartKey().getRow(),
+ r.isStartKeyInclusive(), r.getEndKey() == null ? null : r.getEndKey().getRow())
+ .build();
+ List<KeyExtent> result = new ArrayList<>();
+ for (TabletMetadata t : tm) {
+ LOG.trace("Evaluating: {}", t.getExtent());
+ if (t.getExtent().endRow() == null || r.getEndKey() == null || t.getExtent().endRow() != null
+ && r.getEndKey().compareTo(new Key(t.getExtent().endRow()), PartialKey.ROW) >= 0) {
+ LOG.trace("Adding: {}", t.getExtent());
+ result.add(t.getExtent());
+ }
+ }
+ LOG.trace("Loaded extents: {}", result);
+ return result;
+ }
+
+ public Map<Range,List<KeyExtent>> lookup(TableId tid, List<Range> ranges) {
+ LOG.trace("Performing lookup for table: {}, range: {}", tid, ranges);
+ Map<Range,List<KeyExtent>> results = new TreeMap<>();
+ TreeMap<Range,List<KeyExtent>> extentMap = cache.get(tid, (t) -> {
+ return new TreeMap<Range,List<KeyExtent>>();
+ });
+
+ Collections.sort(ranges);
+ ranges.forEach(r -> {
+ // Find an existing range in the extentMap that overlaps the scan range
+ boolean foundRange = false;
+ Range rangeBefore = extentMap.floorKey(r);
+ if (rangeBefore != null) {
+ Map<Range,List<KeyExtent>> tm = extentMap.tailMap(rangeBefore);
+ for (Entry<Range,List<KeyExtent>> e : tm.entrySet()) {
+ if (e.getKey().overlaps(r)) {
+ // Ignore extents that don't contain the scan range
+ List<KeyExtent> extents = new ArrayList<>();
+ if (r.getStartKey() == null) {
+ extents.addAll(e.getValue());
+ } else {
+ e.getValue().forEach(extent -> {
+ Text row = r.isStartKeyInclusive() ? r.getStartKey().getRow()
+ : r.getStartKey().followingKey(PartialKey.ROW).getRow();
+ if (extent.contains(row)) {
+ extents.add(extent);
+ }
+ });
+ }
+ results.put(r, extents);
+ foundRange = true;
+ break;
Review Comment:
Getting a bit lost w/ the curly braces, but it seems like this takes only the first range in the cache that overlaps the query range. If so, what if there are other ranges in the cache that overlap the query range?
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1331003307
@keith-turner - tests added, kicking 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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1037498364
##########
core/src/main/thrift/tabletserver.thrift:
##########
@@ -265,6 +265,8 @@ service TabletScanClientService {
1:client.ThriftSecurityException sec
2:TSampleNotPresentException tsnpe
3:ScanServerBusyException ssbe
+ 4:NotServingTabletException nste
+ 5:TooManyFilesException tmfe
Review Comment:
Rebased branch to main, so this is OBE now
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
keith-turner commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1031344021
##########
test/src/main/java/org/apache/accumulo/test/ScanServerIT.java:
##########
@@ -148,16 +212,56 @@ public void testBatchScan() throws Exception {
}
@Test
- public void testScanOfflineTable() throws Exception {
+ public void testBatchScanTableCreatedOffline() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+ client.tableOperations().create(tableName, new NewTableConfiguration().createOffline());
+
+ try (BatchScanner scanner = client.createBatchScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRanges(Collections.singletonList(new Range()));
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ } // when the scanner is closed, all open sessions should be closed
Review Comment:
```suggestion
assertEquals(0, Iterables.size(scanner), "Unexpected entries seen when scanning empty table");
} // when the scanner is closed, all open sessions should be closed
assertFalse(client.tableOperations().isOnline(tableName));
```
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchReaderIterator.java:
##########
@@ -241,38 +247,59 @@ private void binRanges(TabletLocator tabletLocator, List<Range> ranges,
int lastFailureSize = Integer.MAX_VALUE;
- while (true) {
+ if (options.getConsistencyLevel().equals(ConsistencyLevel.EVENTUAL) && !this.tableOnline) {
+ // The TabletLocator only caches the location of tablets for online tables. The ScanServer
+ // can scan online and offline tables, but does not require the location of the tablet. It
+ // only requires the KeyExtent. Find the first extent for the table that matches the
+ // startRow provided in the scanState. If no startRow is provided, use the first KeyExtent
+ // for the table
binnedRanges.clear();
- List<Range> failures = tabletLocator.binRanges(context, ranges, binnedRanges);
-
- if (failures.isEmpty()) {
- break;
- } else {
- // tried to only do table state checks when failures.size() == ranges.size(), however this
- // did
- // not work because nothing ever invalidated entries in the tabletLocator cache... so even
- // though
- // the table was deleted the tablet locator entries for the deleted table were not
- // cleared... so
- // need to always do the check when failures occur
- if (failures.size() >= lastFailureSize) {
- context.requireNotDeleted(tableId);
- context.requireNotOffline(tableId, tableName);
+ Map<KeyExtent,List<Range>> map = new HashMap<>();
+ ranges.forEach(r -> {
+ TabletsMetadata tm = context.getAmple().readTablets().forTable(this.tableId)
+ .overlapping(r.getStartKey() == null ? null : r.getStartKey().getRow(),
+ r.isStartKeyInclusive(), r.getEndKey() == null ? null : r.getEndKey().getRow())
+ .build();
Review Comment:
Always doing a metadata lookup per range for the batch scanner could be a performance problem. Would be better to have some sort of caching. Can cache previously seen extents as long as the cache is invalidated when scan server reports back that extent does not exists.
There may be some code from the bulk import code that we could use, for the case when bulk import works with an offline table. I will look around for that.
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftScanner.java:
##########
@@ -288,63 +290,88 @@ public static List<KeyValue> scan(ClientContext context, ScanState scanState, lo
if ((System.currentTimeMillis() - startTime) / 1000.0 > timeOut)
throw new ScanTimedOutException();
- while (loc == null) {
- long currentTime = System.currentTimeMillis();
- if ((currentTime - startTime) / 1000.0 > timeOut)
- throw new ScanTimedOutException();
-
- Span child1 = TraceUtil.startSpan(ThriftScanner.class, "scan::locateTablet");
- try (Scope locateSpan = child1.makeCurrent()) {
- loc = TabletLocator.getLocator(context, scanState.tableId).locateTablet(context,
- scanState.startRow, scanState.skipStartRow, false);
-
- if (loc == null) {
- context.requireNotDeleted(scanState.tableId);
- context.requireNotOffline(scanState.tableId, null);
-
- error = "Failed to locate tablet for table : " + scanState.tableId + " row : "
- + scanState.startRow;
+ boolean tableOnline =
+ context.tableOperations().isOnline(context.getTableName(scanState.tableId));
+
+ if (scanState.runOnScanServer && !tableOnline) {
+ // The TabletLocator only caches the location of tablets for online tables. The ScanServer
+ // can scan online and offline tables, but does not require the location of the tablet. It
+ // only requires the KeyExtent. Find the first extent for the table that matches the
+ // startRow provided in the scanState. If no startRow is provided, use the first KeyExtent
+ // for the table
+ Text endRow = scanState.range != null && scanState.range.getEndKey() != null
+ ? scanState.range.getEndKey().getRow() : null;
+ TabletsMetadata tm = context.getAmple().readTablets().forTable(scanState.tableId)
Review Comment:
Would be better to cache extents.
##########
test/src/main/java/org/apache/accumulo/test/ScanServerIT.java:
##########
@@ -123,21 +127,81 @@ public void testScan() throws Exception {
}
}
+ @Test
+ public void testScanOfflineSingleTablet() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", false);
+ client.tableOperations().offline(tableName, true);
+ assertFalse(client.tableOperations().isOnline(tableName));
+
+ TabletLocator.getLocator((ClientContext) client,
+ TableId.of(client.tableOperations().tableIdMap().get(tableName))).invalidateCache();
+
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "The scan server scanner should have seen all ingested and flushed entries");
+ } // when the scanner is closed, all open sessions should be closed
+ }
+ }
+
+ @Test
+ public void testScanOfflineTableManyTablets() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", true);
+ client.tableOperations().offline(tableName, true);
+ assertFalse(client.tableOperations().isOnline(tableName));
+
+ TabletLocator.getLocator((ClientContext) client,
+ TableId.of(client.tableOperations().tableIdMap().get(tableName))).invalidateCache();
+
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "The scan server scanner should have seen all ingested and flushed entries");
+ } // when the scanner is closed, all open sessions should be closed
+ }
+ }
+
+ @Test
+ public void testScanTableCreatedOffline() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+ client.tableOperations().create(tableName, new NewTableConfiguration().createOffline());
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ } // when the scanner is closed, all open sessions should be closed
Review Comment:
```suggestion
assertEquals(0, Iterables.size(scanner), "Unexpected entries seen when scanning empty table");
} // when the scanner is closed, all open sessions should be closed
assertFalse(client.tableOperations().isOnline(tableName));
```
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1323955190
ReadWriteIT now passing locally, 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 pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1331399241
Full IT build succeeded
--
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] DomGarguilo commented on a diff in pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
DomGarguilo commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1034021294
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/KeyExtentCache.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.clientImpl;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.accumulo.core.util.Pair;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.CacheLoader;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.github.benmanes.caffeine.cache.Scheduler;
+
+public class KeyExtentCache {
+
+ private static class KeyExtentCacheLoader
+ implements CacheLoader<Pair<TableId,Range>,List<KeyExtent>> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KeyExtentCacheLoader.class);
+
+ private final ClientContext ctx;
+
+ public KeyExtentCacheLoader(ClientContext ctx) {
+ this.ctx = ctx;
+ }
+
+ @Override
+ public @Nullable List<KeyExtent> load(Pair<TableId,Range> pair) throws Exception {
+ final TableId tid = pair.getFirst();
+ final Range r = pair.getSecond();
+ LOG.trace("Loading extents for tid: {} and range: {}", tid, r);
+ TabletsMetadata tm = ctx.getAmple().readTablets().forTable(tid)
+ .overlapping(r.getStartKey() == null ? null : r.getStartKey().getRow(),
+ r.isStartKeyInclusive(), r.getEndKey() == null ? null : r.getEndKey().getRow())
+ .build();
+ List<KeyExtent> result = new ArrayList<>();
+ for (TabletMetadata t : tm) {
+ if (t.getExtent().endRow() == null || t.getExtent().endRow() != null
+ && !r.afterEndKey(new Key(t.getExtent().endRow()).followingKey(PartialKey.ROW))) {
+ result.add(t.getExtent());
+ }
+ }
+ LOG.trace("Returning extents: {}", result);
+ return result;
+ }
+
+ @Override
+ public Map<? extends Pair<TableId,Range>,? extends List<KeyExtent>>
+ loadAll(Set<? extends Pair<TableId,Range>> pairs) throws Exception {
+
+ // Sort the inputs
+ TreeMap<TableId,SortedSet<Range>> inputs = new TreeMap<>();
Review Comment:
You might be able to use a TreeMultimap here. Not sure if it would make much of a difference though.
--
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 pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
keith-turner commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1329783854
> Regarding this - how would the client know whether the cache contents are correct or not if it doesn't query the metadata table each time to compare the cached data?
Thinking the way it would work is that if a tablet splits and no longer exist, but is still in a clients cache, that it would eventually go to a scan server that would report the tablet does not exists and it would then clear its cache entry for that tablet. After its cleared it would reload it eventually. I think this is how caching works w/ tservers. It stays in the cache until a tserver reports back it does not exists. I think this was working for scan servers too when they were querying online tables, it would take the same code path when a scan server or tserver reported a tablet no longer existed.
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1355610152
This is on pause while @keith-turner looks at modifying TabletLocator to cache extents for offline tables.
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1029394099
##########
test/src/main/java/org/apache/accumulo/test/ScanServerNoSServersIT.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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;
+
+import static org.apache.accumulo.harness.AccumuloITBase.MINI_CLUSTER_ONLY;
+import static org.apache.accumulo.test.ScanServerIT.createTableAndIngest;
+import static org.apache.accumulo.test.ScanServerIT.ingest;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Collections;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.ScannerBase.ConsistencyLevel;
+import org.apache.accumulo.core.client.TableOfflineException;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+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.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import com.google.common.collect.Iterables;
+
+@Tag(MINI_CLUSTER_ONLY)
+public class ScanServerNoSServersIT extends SharedMiniClusterBase {
+
+ // This is the same as ScanServerIT, but without any scan servers running.
+ // This tests the cases where the client falls back to the Tablet Servers
+
+ private static class ScanServerITConfiguration implements MiniClusterConfigurationCallback {
+
+ @Override
+ public void configureMiniCluster(MiniAccumuloConfigImpl cfg,
+ org.apache.hadoop.conf.Configuration coreSite) {
+ cfg.setNumScanServers(0);
+
+ // Timeout scan sessions after being idle for 3 seconds
+ cfg.setProperty(Property.TSERV_SESSION_MAXIDLE, "3s");
+
+ // Configure the scan server to only have 1 scan executor thread. This means
+ // that the scan server will run scans serially, not concurrently.
+ cfg.setProperty(Property.SSERV_SCAN_EXECUTORS_DEFAULT_THREADS, "1");
+ }
+ }
+
+ @BeforeAll
+ public static void start() throws Exception {
+ ScanServerITConfiguration c = new ScanServerITConfiguration();
+ SharedMiniClusterBase.startMiniClusterWithConfig(c);
+ }
+
+ @AfterAll
+ public static void stop() throws Exception {
+ SharedMiniClusterBase.stopMiniCluster();
+ }
+
+ @Test
+ public void testScan() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", false);
+
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "Scanner did not see ingested and flushed entries");
+ final int additionalIngestedEntryCount =
+ ingest(client, tableName, 10, 10, 10, "colf", false, 1);
+ // since there are no scan servers, and we are reading from tservers, we should see update
+ assertEquals(ingestedEntryCount + additionalIngestedEntryCount, Iterables.size(scanner),
+ "Scanning against tserver should have resulted in seeing all ingested entries");
+ } // when the scanner is closed, all open sessions should be closed
+ }
+ }
+
+ @Test
+ public void testScanOfflineSingleTablet() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", false);
+ client.tableOperations().offline(tableName, true);
+ assertFalse(client.tableOperations().isOnline(tableName));
+
+ TabletLocator.getLocator((ClientContext) client,
+ TableId.of(client.tableOperations().tableIdMap().get(tableName))).invalidateCache();
+
+ assertThrows(TableOfflineException.class, () -> {
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "The scan server scanner should have seen all ingested and flushed entries");
+ } // when the scanner is closed, all open sessions should be closed
+ });
+ }
+ }
+
+ @Test
+ public void testScanOfflineTableManyTablets() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+
+ final int ingestedEntryCount =
+ createTableAndIngest(client, tableName, null, 10, 10, "colf", true);
+ client.tableOperations().offline(tableName, true);
+ assertFalse(client.tableOperations().isOnline(tableName));
+
+ TabletLocator.getLocator((ClientContext) client,
+ TableId.of(client.tableOperations().tableIdMap().get(tableName))).invalidateCache();
+
+ assertThrows(TableOfflineException.class, () -> {
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.setConsistencyLevel(ConsistencyLevel.EVENTUAL);
+ assertEquals(ingestedEntryCount, Iterables.size(scanner),
+ "The scan server scanner should have seen all ingested and flushed entries");
+ } // when the scanner is closed, all open sessions should be closed
+ });
+ }
+ }
+
+ @Test
+ public void testScanTableCreatedOffline() throws Exception {
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+ client.tableOperations().create(tableName, new NewTableConfiguration().createOffline());
+ try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
Review Comment:
I didn't try to read from the scanner, when this is done, it throws the TableOfflineException as it should.
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
keith-turner commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1035899411
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/OfflineTableExtentCache.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.clientImpl;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.accumulo.core.util.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.CacheLoader;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.github.benmanes.caffeine.cache.Scheduler;
+
+public class OfflineTableExtentCache {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OfflineTableExtentCache.class);
+
+ private static class KeyExtentCacheLoader
Review Comment:
Other caches for tablets in Accumulo use TreeMap, which allows finding overlapping ranges.
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#issuecomment-1540054906
We are no longer planning on allowing eventual scans on offline tables
--
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 #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
dlmarion commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1039894237
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/OfflineTableExtentCache.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.clientImpl;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.accumulo.core.util.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.CacheLoader;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.github.benmanes.caffeine.cache.Scheduler;
+
+public class OfflineTableExtentCache {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OfflineTableExtentCache.class);
+
+ private static class KeyExtentCacheLoader
Review Comment:
@keith-turner - I rewrote the OfflineTableExtentCache, it should find extents for the current scan based on cached ranges and extents from prior scans.
--
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] ctubbsii commented on a diff in pull request #3091: Added code and tests to confirm ScanServers can scan offline tables
Posted by GitBox <gi...@apache.org>.
ctubbsii commented on code in PR #3091:
URL: https://github.com/apache/accumulo/pull/3091#discussion_r1041807613
##########
core/src/main/java/org/apache/accumulo/core/data/RangedObject.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.data;
+
+/**
+ * Marker interface that declares the object is a type that represents a range
+ */
+public interface RangedObject<T> {
Review Comment:
I'm not sure this interface makes sense to add. Although it's helpful to have a similar method on the `Range` class and `KeyExtent` because they serve similar roles in our public API and internal implementation, respectively, they are not closely related enough to require an interface to try to make them related. The need to use a generic type for its argument reveals how distantly related they are, as well as the fact that they have independent implementations, and we never assign either to any variable of the interface type for any purpose.
I think it is sufficient to add the `overlaps` method to `Range` (with a corresponding `@since` tag on the new method) without adding this interface to the public API.
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/ScannerImpl.java:
##########
@@ -152,6 +153,16 @@ public synchronized int getBatchSize() {
@Override
public synchronized Iterator<Entry<Key,Value>> iterator() {
ensureOpen();
+
+ if (getConsistencyLevel() == ConsistencyLevel.IMMEDIATE) {
+ try {
+ String tableName = context.getTableName(tableId);
+ context.requireNotOffline(tableId, tableName);
+ } catch (TableNotFoundException e) {
+ throw new RuntimeException("Table not found", e);
Review Comment:
I think we probably need a generic `UncheckedAccumuloException` so we can avoid the base RTE class and be able to easily distinguish from our RTEs and ones that libraries/user code might throw. `UncheckedIOException` is a good model to follow (though we don't always utilize that when we can; not currently, anyway).
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftScanner.java:
##########
@@ -559,6 +580,11 @@ public Map<String,String> getHints() {
"For tablet {} scan server selector chose scan_server:{} delay:{} busyTimeout:{}",
loc.tablet_extent, scanServer, delay, scanState.busyTimeout);
} else {
+ try {
+ context.requireNotOffline(scanState.tableId, context.getTableName(scanState.tableId));
+ } catch (TableNotFoundException e) {
+ throw new RuntimeException("Error scanning for table: " + scanState.tableId, e);
Review Comment:
Some of these generic RTEs might be better expressed as IllegalStateException. In general, it's preferable to use a more specific RTE whenever possible.
--
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