fix major issues from sonarqube

NationalSecurityAgency · Apr 5, 2024 · dd6a584 · dd6a584
1 parent a4587af
commit dd6a584
Show file tree

Hide file tree

Showing 74 changed files with 247 additions and 79 deletions.
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
@@ -416,6 +416,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
@@ -903,6 +903,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;

diff --git a/...query-core/src/main/java/datawave/core/iterators/DatawaveFieldIndexRegexIteratorJexl.java b/...query-core/src/main/java/datawave/core/iterators/DatawaveFieldIndexRegexIteratorJexl.java
@@ -4,15 +4,13 @@
 import java.util.List;
 import java.util.regex.Pattern;
 
-import org.apache.accumulo.core.data.Column;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.hadoop.io.Text;
 
-import datawave.data.ColumnFamilyConstants;
 import datawave.query.Constants;
 import datawave.query.parser.JavaRegexAnalyzer;
 import datawave.query.parser.JavaRegexAnalyzer.JavaRegexParseException;

diff --git a/warehouse/query-core/src/main/java/datawave/mr/bulk/MultiRfileInputformat.java b/warehouse/query-core/src/main/java/datawave/mr/bulk/MultiRfileInputformat.java
@@ -94,7 +94,10 @@ public List<InputSplit> getSplits(JobContext job) throws IOException {
         List<InputSplit> inputSplits = Lists.newArrayList();
         try {
             inputSplits = computeSplitPoints(job, tableName, ranges);
-        } catch (TableNotFoundException | AccumuloException | AccumuloSecurityException | InterruptedException e) {
+        } catch (TableNotFoundException | AccumuloException | AccumuloSecurityException e) {
+            throw new IOException(e);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             throw new IOException(e);
         }
 

diff --git a/warehouse/query-core/src/main/java/datawave/mr/bulk/RecordIterator.java b/warehouse/query-core/src/main/java/datawave/mr/bulk/RecordIterator.java
@@ -347,6 +347,8 @@ protected synchronized void initialize(Configuration conf, boolean initRanges) t
             } catch (ExecutionException e) {
                 close();
                 throw new RuntimeException(e.getCause());
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
             } catch (Exception e) {
 
             }
@@ -562,6 +564,7 @@ protected synchronized void fail(Throwable t) {
         try {
             Thread.sleep(failureSleep);
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             try {
                 close();
             } catch (IOException e1) {
@@ -590,6 +593,11 @@ protected synchronized void fail(Throwable t) {
 
             seekLastSeen();
 
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            if (!callClosed.get() && !Thread.interrupted()) {
+                fail(e);
+            }
         } catch (Exception e) {
             if (!callClosed.get() && !Thread.interrupted())
                 fail(e);
@@ -853,7 +861,7 @@ public synchronized void setBlockFile(Reader reader) throws InterruptedException
             this.reader = reader;
         }
 
-        public FSDataInputStream getInputStream() {
+        public synchronized FSDataInputStream getInputStream() {
             return inputStream;
         }
 

diff --git a/warehouse/query-core/src/main/java/datawave/mr/bulk/RfileScanner.java b/warehouse/query-core/src/main/java/datawave/mr/bulk/RfileScanner.java
@@ -203,7 +203,10 @@ public Iterator<Entry<Key,Value>> iterator() {
                         log.trace("KV iterator is " + (kv == null ? "null" : "not null"));
                     }
                 } while (null == kv);
-
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                IOUtils.cleanupWithLogger(null, this);
+                throw new RuntimeException(e);
             } catch (Exception e) {
                 IOUtils.cleanupWithLogger(null, this);
                 throw new RuntimeException(e);