NationalSecurityAgency · cogross · Apr 1, 2024
diff --git a/core/connection-pool/src/main/java/datawave/core/common/cache/SharedCacheCoordinator.java b/core/connection-pool/src/main/java/datawave/core/common/cache/SharedCacheCoordinator.java
@@ -726,11 +726,17 @@ protected void reapEvictions() {
                         String recursiveDeletePath = ZKPaths.makePath(curatorClient.getNamespace(), path);
                         ZKUtil.deleteRecursive(curatorClient.getZookeeperClient().getZooKeeper(), recursiveDeletePath);
                     } catch (Exception e) {
+                        if (e instanceof InterruptedException) {
+                            Thread.currentThread().interrupt();
+                        }
                         log.trace("Problem deleting " + path + " (this may be ok): " + e.getMessage(), e);
                     }
                 }
             }
         } catch (Exception e) {
+            if (e instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+            }
             log.warn("Error cleaning up eviction notices: " + e.getMessage(), e);
         }
     }

diff --git a/...ce/src/main/java/datawave/core/mapreduce/bulkresults/map/BulkResultsFileOutputMapper.java b/...ce/src/main/java/datawave/core/mapreduce/bulkresults/map/BulkResultsFileOutputMapper.java
@@ -119,6 +119,9 @@ protected void map(Key key, Value value, org.apache.hadoop.mapreduce.Mapper<Key,
                     }
                     context.write(key, val);
                 } catch (Exception e) {
+                    if (e instanceof InterruptedException) {
+                        Thread.currentThread().interrupt();
+                    }
                     throw new RuntimeException("Unable to serialize response of class: " + response.getClass().getName(), e);
                 }
                 context.progress();

diff --git a/...e/src/main/java/datawave/core/mapreduce/bulkresults/map/BulkResultsTableOutputMapper.java b/...e/src/main/java/datawave/core/mapreduce/bulkresults/map/BulkResultsTableOutputMapper.java
@@ -89,6 +89,9 @@ protected void map(Key key, Value value, org.apache.hadoop.mapreduce.Mapper<Key,
                     m.put(key.getColumnFamily(), key.getColumnQualifier(), new ColumnVisibility(key.getColumnVisibility()), key.getTimestamp(), val);
                     context.write(this.tableName, m);
                 } catch (Exception e) {
+                    if (e instanceof InterruptedException) {
+                        Thread.currentThread().interrupt();
+                    }
                     throw new RuntimeException("Unable to serialize response of class: " + response.getClass().getName(), e);
                 }
                 context.progress();

diff --git a/core/query/src/main/java/datawave/core/query/logic/composite/CompositeQueryLogic.java b/core/query/src/main/java/datawave/core/query/logic/composite/CompositeQueryLogic.java
@@ -150,6 +150,7 @@ public void run() {
                             log.debug(Thread.currentThread().getName() + ": Got null result");
                         }
                     } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
                         // if this was on purpose, then just log and the loop will naturally exit
                         if (interrupted) {
                             log.warn("QueryLogic thread interrupted", e);
@@ -478,6 +479,7 @@ public void close() {
             try {
                 holder.join();
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                 log.error("Error joining query logic thread", e);
                 throw new RuntimeException("Error joining query logic thread", e);
             }

diff --git a/...src/main/java/datawave/core/query/logic/composite/CompositeQueryLogicResultsIterator.java b/...src/main/java/datawave/core/query/logic/composite/CompositeQueryLogicResultsIterator.java
@@ -68,6 +68,7 @@ public boolean hasNext() {
                     if (failure != null) {
                         Throwables.propagate(failure);
                     }
+                    Thread.currentThread().interrupt();
                     throw new RuntimeException(e);
                 }
             }

diff --git a/...ervice/src/main/java/datawave/webservice/mr/input/SecureEventSequenceFileInputFormat.java b/...ervice/src/main/java/datawave/webservice/mr/input/SecureEventSequenceFileInputFormat.java
@@ -17,6 +17,7 @@ public RecordReader<K,RawRecordContainer> createRecordReader(InputSplit split, T
         try {
             reader.initialize(split, context);
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             throw new IOException("Error initializing SecureEventSequenceFileRecordReader", e);
         }
         return reader;

diff --git a/warehouse/ingest-core/src/main/java/datawave/ingest/data/tokenize/TokenizationHelper.java b/warehouse/ingest-core/src/main/java/datawave/ingest/data/tokenize/TokenizationHelper.java
@@ -1,6 +1,7 @@
 package datawave.ingest.data.tokenize;
 
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.log4j.Logger;
@@ -25,6 +26,7 @@ public static class HeartBeatThread extends Thread {
 
         public static final long INTERVAL = 500; // half second resolution
         public static volatile int counter = 0;
+
         public static long lastRun;
 
         static {
@@ -41,6 +43,7 @@ public void run() {
                 try {
                     Thread.sleep(INTERVAL);
                 } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
                     throw new RuntimeException(e);
                 }
 

diff --git a/...use/ingest-core/src/main/java/datawave/ingest/input/reader/AbstractEventRecordReader.java b/...use/ingest-core/src/main/java/datawave/ingest/input/reader/AbstractEventRecordReader.java
@@ -175,6 +175,9 @@ public RawRecordContainer getEvent() {
 
         try {
             event.setRawRecordNumber(getCurrentKey().get());
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException("Unable to get current key", e);
         } catch (Exception e) {
             throw new RuntimeException("Unable to get current key", e);
         }

diff --git a/warehouse/ingest-core/src/main/java/datawave/ingest/mapreduce/EventMapper.java b/warehouse/ingest-core/src/main/java/datawave/ingest/mapreduce/EventMapper.java
@@ -253,6 +253,9 @@ public void setup(Context context) throws IOException, InterruptedException {
         try {
             contextWriter = contextWriterClass.getDeclaredConstructor().newInstance();
             contextWriter.setup(filterConf, filterConf.getBoolean(CONTEXT_WRITER_OUTPUT_TABLE_COUNTERS, false));
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("Failed to initialized " + contextWriterClass + " from property " + CONTEXT_WRITER_CLASS, e);
         } catch (Exception e) {
             throw new IOException("Failed to initialized " + contextWriterClass + " from property " + CONTEXT_WRITER_CLASS, e);
         }
@@ -480,7 +483,11 @@ public void map(K1 key, V1 value, Context context) throws IOException, Interrupt
                     NDC.push(origFiles.iterator().next());
                     reprocessedNDCPush = true;
                 }
-
+            } catch (InterruptedException e) {
+                contextWriter.rollback();
+                Thread.currentThread().interrupt();
+                log.error("Failed to clean event from error table.  Terminating map", e);
+                throw new IOException("Failed to clean event from error table, Terminating map", e);
             } catch (Exception e) {
                 contextWriter.rollback();
                 log.error("Failed to clean event from error table.  Terminating map", e);

diff --git a/.../ingest-core/src/main/java/datawave/ingest/mapreduce/handler/facet/HashTableFunction.java b/.../ingest-core/src/main/java/datawave/ingest/mapreduce/handler/facet/HashTableFunction.java
@@ -107,7 +107,10 @@ public Collection<NormalizedContentInterface> apply(@Nullable Collection<Normali
 
             try {
                 contextWriter.write(map, context);
-            } catch (IOException | InterruptedException e) {
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException(e);
+            } catch (IOException e) {
                 throw new RuntimeException(e);
             }
 

diff --git a/...in/java/datawave/ingest/mapreduce/handler/tokenize/ContentIndexingColumnBasedHandler.java b/...in/java/datawave/ingest/mapreduce/handler/tokenize/ContentIndexingColumnBasedHandler.java
@@ -195,7 +195,7 @@ protected void flushTokenOffsetCache(RawRecordContainer event, Multimap<BulkInge
                 log.fatal("IOException", ex);
             } catch (InterruptedException ex) {
                 log.warn("Interrupted!", ex);
-                Thread.interrupted();
+                Thread.currentThread().interrupt();
             }
 
             tokenOffsetCache.clear();
@@ -303,6 +303,9 @@ protected Multimap<String,NormalizedContentInterface> getShardNamesAndValues(Raw
                 if (indexField || reverseIndexField) {
                     try {
                         tokenizeField(analyzer, nci, indexField, reverseIndexField, reporter);
+                    } catch (InterruptedException ex) {
+                        Thread.currentThread().interrupt();
+                        throw new RuntimeException(ex);
                     } catch (Exception ex) {
                         throw new RuntimeException(ex);
                     }

diff --git a/...datawave/ingest/mapreduce/handler/tokenize/ExtendedContentIndexingColumnBasedHandler.java b/...datawave/ingest/mapreduce/handler/tokenize/ExtendedContentIndexingColumnBasedHandler.java
@@ -229,7 +229,10 @@ public void close(TaskAttemptContext context) {
                 this.docWriterService.shutdown();
                 this.docWriterService.awaitTermination(1, TimeUnit.MINUTES);
                 this.docWriter.close();
-            } catch (InterruptedException | MutationsRejectedException e) {
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.error("Unable to terminate document writing service!", e);
+            } catch (MutationsRejectedException e) {
                 log.error("Unable to terminate document writing service!", e);
             }
         }
@@ -669,6 +672,7 @@ public void run() {
                 try {
                     Thread.sleep(INTERVAL);
                 } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
                     throw new RuntimeException(e);
                 }
 

diff --git a/...ouse/ingest-core/src/main/java/datawave/ingest/mapreduce/job/BulkIngestMapFileLoader.java b/...ouse/ingest-core/src/main/java/datawave/ingest/mapreduce/job/BulkIngestMapFileLoader.java
@@ -512,6 +512,7 @@ public void run() {
                                             try {
                                                 Thread.sleep(FAILURE_SLEEP_TIME);
                                             } catch (InterruptedException ie) {
+                                                Thread.currentThread().interrupt();
                                                 log.warn("Interrupted while sleeping.", ie);
                                             }
                                         }
@@ -822,6 +823,7 @@ public void bringMapFilesOnline(Path mapFilesDir) throws IOException, AccumuloEx
                         if (e == null)
                             e = importTask.getException();
                     } catch (InterruptedException interrupted) {
+                        Thread.currentThread().interrupt();
                         // this task was interrupted, wait for the others and then terminate
                         if (e == null)
                             e = interrupted;
@@ -844,6 +846,7 @@ public void bringMapFilesOnline(Path mapFilesDir) throws IOException, AccumuloEx
                 if (e == null)
                     e = importTask.getException();
             } catch (InterruptedException interrupted) {
+                Thread.currentThread().interrupt();
                 // this task was interrupted, wait for the others and then terminate
                 if (e == null)
                     e = interrupted;
@@ -1238,6 +1241,7 @@ public void markSourceFilesLoaded(Path jobDirectory) throws IOException {
             }
 
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             if (null != e.getCause())
                 throw new IOException(e.getCause().getMessage());
             else
@@ -1320,6 +1324,7 @@ private void sleep() {
             System.gc();
             Thread.sleep(SLEEP_TIME);
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             log.warn("Interrupted while sleeping.", e);
         }
     }

diff --git a/warehouse/ingest-core/src/main/java/datawave/ingest/mapreduce/job/IngestJob.java b/warehouse/ingest-core/src/main/java/datawave/ingest/mapreduce/job/IngestJob.java
@@ -415,6 +415,7 @@ public int run(String[] args) throws Exception {
                 try {
                     Thread.sleep(3000);
                 } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
                     // do nothing
                 }
             }

diff --git a/...est-core/src/main/java/datawave/ingest/mapreduce/job/metrics/AggregatingMetricsStore.java b/...est-core/src/main/java/datawave/ingest/mapreduce/job/metrics/AggregatingMetricsStore.java
@@ -60,6 +60,9 @@ public void flush(ConcurrentMap<String,AtomicLong> counts) {
                     byte[] countAsBytes = TextUtil.toUtf8(String.valueOf(entry.getValue().get()));
                     Key key = KeyConverter.fromString(entry.getKey());
                     contextWriter.write(new BulkIngestKey(table, key), new Value(countAsBytes), context);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    logger.error("Could not flush metrics, dropping them", e);
                 } catch (Exception e) {
                     logger.error("Could not flush metrics, dropping them", e);
                 }

diff --git a/...est-core/src/main/java/datawave/ingest/mapreduce/job/metrics/PassThroughMetricsStore.java b/...est-core/src/main/java/datawave/ingest/mapreduce/job/metrics/PassThroughMetricsStore.java
@@ -47,6 +47,9 @@ public void increase(String key, long count) {
             Value v = (count == 1L && ONE != null) ? ONE : new Value(String.valueOf(count).getBytes(ENCODING));
             Key k = KeyConverter.fromString(key);
             contextWriter.write(new BulkIngestKey(table, k), v, context);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            logger.error("Could not write metrics to the context writer, dropping them...", e);
         } catch (Exception e) {
             logger.error("Could not write metrics to the context writer, dropping them...", e);
         }

diff --git a/...e/src/main/java/datawave/ingest/mapreduce/job/reduce/BulkIngestKeyAggregatingReducer.java b/...e/src/main/java/datawave/ingest/mapreduce/job/reduce/BulkIngestKeyAggregatingReducer.java
@@ -65,6 +65,9 @@ protected void setupContextWriter(Configuration conf) throws IOException {
         try {
             setContextWriter(contextWriterClass.newInstance());
             contextWriter.setup(conf, conf.getBoolean(CONTEXT_WRITER_OUTPUT_TABLE_COUNTERS, false));
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("Failed to initialized " + contextWriterClass + " from property " + CONTEXT_WRITER_CLASS, e);
         } catch (Exception e) {
             throw new IOException("Failed to initialized " + contextWriterClass + " from property " + CONTEXT_WRITER_CLASS, e);
         }

diff --git a/...-core/src/main/java/datawave/ingest/mapreduce/job/reduce/BulkIngestKeyDedupeCombiner.java b/...-core/src/main/java/datawave/ingest/mapreduce/job/reduce/BulkIngestKeyDedupeCombiner.java
@@ -56,6 +56,9 @@ protected void setupContextWriter(Configuration conf) throws IOException {
         try {
             setContextWriter(contextWriterClass.getDeclaredConstructor().newInstance());
             contextWriter.setup(conf, conf.getBoolean(CONTEXT_WRITER_OUTPUT_TABLE_COUNTERS, false));
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("Failed to initialized " + contextWriterClass + " from property " + CONTEXT_WRITER_CLASS, e);
         } catch (Exception e) {
             throw new IOException("Failed to initialized " + contextWriterClass + " from property " + CONTEXT_WRITER_CLASS, e);
         }

diff --git a/...core/src/main/java/datawave/ingest/mapreduce/job/writer/AbstractChainedContextWriter.java b/...core/src/main/java/datawave/ingest/mapreduce/job/writer/AbstractChainedContextWriter.java
@@ -44,6 +44,9 @@ public void setup(Configuration conf, boolean outputTableCounters) throws IOExce
         try {
             contextWriter = contextWriterClass.newInstance();
             contextWriter.setup(conf, outputTableCounters);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("Failed to initialized " + contextWriterClass + " from property " + getChainedContextWriterOption(), e);
         } catch (Exception e) {
             throw new IOException("Failed to initialized " + contextWriterClass + " from property " + getChainedContextWriterOption(), e);
         }

diff --git a/...st-core/src/main/java/datawave/ingest/mapreduce/job/writer/TableCachingContextWriter.java b/...st-core/src/main/java/datawave/ingest/mapreduce/job/writer/TableCachingContextWriter.java
@@ -90,6 +90,9 @@ public void setup(Configuration conf, boolean outputTableCounters) throws IOExce
         try {
             contextWriter = contextWriterClass.getDeclaredConstructor().newInstance();
             contextWriter.setup(conf, outputTableCounters);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("Failed to initialized " + contextWriterClass + " from property " + CONTEXT_WRITER_CLASS, e);
         } catch (Exception e) {
             throw new IOException("Failed to initialized " + contextWriterClass + " from property " + CONTEXT_WRITER_CLASS, e);
         }

diff --git a/warehouse/ingest-core/src/main/java/datawave/ingest/util/KillJobByRegex.java b/warehouse/ingest-core/src/main/java/datawave/ingest/util/KillJobByRegex.java
@@ -38,6 +38,11 @@ public void run() {
             Job job = null;
             try {
                 job = cluster.getJob(js.getJobID());
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                System.out.println("Error getting job: " + js.getJobID() + " Reason: " + e.getMessage());
+                JOB_FAILED_COUNT.getAndIncrement();
+                return;
             } catch (Exception e) {
                 System.out.println("Error getting job: " + js.getJobID() + " Reason: " + e.getMessage());
                 JOB_FAILED_COUNT.getAndIncrement();
@@ -82,6 +87,7 @@ public static void main(String[] args) throws IOException, InterruptedException
             JOB_KILLER_SVC.shutdown(); // signal shutdown
             JOB_KILLER_SVC.awaitTermination(1, TimeUnit.MINUTES); // allow processes to stop
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             JOB_KILLER_SVC.shutdownNow();
         }
 

diff --git a/warehouse/ingest-core/src/main/java/datawave/ingest/util/ThreadUtil.java b/warehouse/ingest-core/src/main/java/datawave/ingest/util/ThreadUtil.java
@@ -29,6 +29,7 @@ public static boolean shutdownAndWait(ThreadPoolExecutor executor, long timeToWa
             executor.awaitTermination(timeToWait, unit);
             return true;
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             logger.warn("Closed thread pool but not all threads completed successfully.");
             return false;
         }

diff --git a/warehouse/ingest-core/src/main/java/datawave/util/flag/FlagMaker.java b/warehouse/ingest-core/src/main/java/datawave/util/flag/FlagMaker.java
@@ -226,6 +226,7 @@ public void run() {
                 } catch (Exception ex) {
                     log.error("An unexpected exception occurred. Exiting", ex);
                     running = false;
+                    Thread.currentThread().interrupt();
                 }
             }
         } finally {
@@ -575,7 +576,10 @@ private boolean processResults(List<Future<InputFile>> futures, Collection<Input
                 if (fe != null && fe.isMoved()) {
                     entries.add(fe);
                 }
-            } catch (InterruptedException | ExecutionException ex) {
+            } catch (InterruptedException ex) {
+                ioex = new IOException("Failure during move", ex.getCause());
+                Thread.currentThread().interrupt();
+            } catch (ExecutionException ex) {
                 ioex = new IOException("Failure during move", ex.getCause());
             }
         }

diff --git a/warehouse/ingest-csv/src/main/java/datawave/ingest/csv/mr/input/CSVReaderBase.java b/warehouse/ingest-csv/src/main/java/datawave/ingest/csv/mr/input/CSVReaderBase.java
@@ -104,6 +104,7 @@ public void initializeTotalSize(final InputSplit genericSplit) throws IOExceptio
         try {
             totalSize = genericSplit.getLength() * 4l;
         } catch (InterruptedException ex) {
+            Thread.currentThread().interrupt();
             throw new IOException("Interrupted Exception thrown while attempting to get split length", ex);
         }
     }

diff --git a/...se/ingest-wikipedia/src/main/java/datawave/ingest/wikipedia/WikipediaDataTypeHandler.java b/...se/ingest-wikipedia/src/main/java/datawave/ingest/wikipedia/WikipediaDataTypeHandler.java
@@ -295,6 +295,10 @@ protected long tokenizeTextNode(String content, RawRecordContainer event, byte[]
                 getMetadata().addEvent(this.ingestHelper, event, normMap);
             }
 
+        } catch (InterruptedException e) {
+            log.error("Error processing Wikipedia document", e);
+            Thread.currentThread().interrupt();
+            throw new RuntimeException("Error processing Wikipedia document", e);
         } catch (Exception e) {
             // If error, return empty results map.
             log.error("Error processing Wikipedia document", e);

diff --git a/warehouse/metrics-core/src/main/java/datawave/metrics/mapreduce/MetricsIngester.java b/warehouse/metrics-core/src/main/java/datawave/metrics/mapreduce/MetricsIngester.java
@@ -101,6 +101,9 @@ public int run(String[] args) throws Exception {
         if ("errors".equals(type)) {
             try {
                 launchErrorsJob(Job.getInstance(conf), conf);
+            } catch (InterruptedException e) {
+                log.info("Failed to launch errors job", e);
+                Thread.currentThread().interrupt();
             } catch (Exception e) {
                 log.info("Failed to launch errors job", e);
             }

diff --git a/warehouse/metrics-core/src/main/java/datawave/metrics/mapreduce/util/JobSetupUtil.java b/warehouse/metrics-core/src/main/java/datawave/metrics/mapreduce/util/JobSetupUtil.java
@@ -113,6 +113,9 @@ public static void changeJobPriority(Job job, Logger log) throws IOException {
             } else {
                 log.error("Hadoop process exited abnormally-- job may take a long time if system is saturated.");
             }
+        } catch (InterruptedException e) {
+            log.error("This job was interrupted.", e);
+            Thread.currentThread().interrupt();
         } catch (Exception e) {
             log.error("This job may take a while on a system running at full ingest load.", e);
         }

diff --git a/...ery-core/src/main/java/datawave/core/iterators/DatawaveFieldIndexCachingIteratorJexl.java b/...ery-core/src/main/java/datawave/core/iterators/DatawaveFieldIndexCachingIteratorJexl.java
@@ -902,6 +902,10 @@ private void fillSortedSets() throws IOException {
             } else {
                 try {
                     result = future.get();
+                } catch (InterruptedException e) {
+                    exception = e;
+                    result = e;
+                    Thread.currentThread().interrupt();
                 } catch (Exception e) {
                     exception = e;
                     result = e;