[accumulo] branch main updated: Fix bug with concurrent merge and bulk import (#3014)

[ct...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ctubbsii pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git


The following commit(s) were added to refs/heads/main by this push:
     new f6c943e70d Fix bug with concurrent merge and bulk import (#3014)
f6c943e70d is described below

commit f6c943e70dd14183985faeeefc5b554fcc731fff
Author: Keith Turner <kt...@apache.org>
AuthorDate: Fri Oct 14 14:41:30 2022 +0100

    Fix bug with concurrent merge and bulk import (#3014)
    
    This fixes #3006
---
 .../clientImpl/bulk/ConcurrentKeyExtentCache.java  | 25 +++++++++++++++++-----
 1 file changed, 20 insertions(+), 5 deletions(-)

diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/ConcurrentKeyExtentCache.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/ConcurrentKeyExtentCache.java
index 03d9bda441..ec154e7dbe 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/ConcurrentKeyExtentCache.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/ConcurrentKeyExtentCache.java
@@ -35,14 +35,19 @@ import org.apache.accumulo.core.clientImpl.ClientContext;
 import org.apache.accumulo.core.clientImpl.bulk.BulkImport.KeyExtentCache;
 import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.TabletDeletedException;
 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.google.common.annotations.VisibleForTesting;
 
 class ConcurrentKeyExtentCache implements KeyExtentCache {
 
+  private static Logger log = LoggerFactory.getLogger(ConcurrentKeyExtentCache.class);
+
   private static final Text MAX = new Text();
 
   private Set<Text> rowsToLookup = Collections.synchronizedSet(new HashSet<>());
@@ -120,12 +125,22 @@ class ConcurrentKeyExtentCache implements KeyExtentCache {
 
         for (Text lookupRow : lookupRows) {
           if (getFromCache(lookupRow) == null) {
-            Iterator<KeyExtent> iter = lookupExtents(lookupRow).iterator();
-            while (iter.hasNext()) {
-              KeyExtent ke2 = iter.next();
-              if (inCache(ke2))
+            while (true) {
+              try {
+                Iterator<KeyExtent> iter = lookupExtents(lookupRow).iterator();
+                while (iter.hasNext()) {
+                  KeyExtent ke2 = iter.next();
+                  if (inCache(ke2))
+                    break;
+                  updateCache(ke2);
+                }
                 break;
-              updateCache(ke2);
+              } catch (TabletDeletedException tde) {
+                // tablets were merged away in the table, start over and try again
+                log.debug("While trying to obtain a tablet location for bulk import, a tablet was "
+                    + "deleted. If this was caused by a concurrent merge tablet "
+                    + "operation, this is okay. Otherwise, it could be a problem.", tde);
+              }
             }
           }
         }