Enables SharedVariableAtomicityDetector

core/pom.xml

@@ -47,6 +47,10 @@
       <artifactId>auto-service</artifactId>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>com.google.code.findbugs</groupId>
+      <artifactId>jsr305</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.code.gson</groupId>
       <artifactId>gson</artifactId>

core/src/main/java/org/apache/accumulo/core/client/ClientSideIteratorScanner.java

@@ -177,6 +177,7 @@ public ClientSideIteratorScanner(final Scanner scanner) {
     this.range = scanner.getRange();
     this.size = scanner.getBatchSize();
     this.retryTimeout = scanner.getTimeout(MILLISECONDS);
+    // TODO Is this intended to be getBatchTimeout()?
     this.batchTimeout = scanner.getTimeout(MILLISECONDS);
     this.readaheadThreshold = scanner.getReadaheadThreshold();
     SamplerConfiguration samplerConfig = scanner.getSamplerConfiguration();

core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java

@@ -28,6 +28,8 @@
 import java.util.Set;
 import java.util.SortedSet;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.client.Scanner;
 import org.apache.accumulo.core.client.TableNotFoundException;
@@ -76,6 +78,7 @@
 
 import com.google.common.base.Preconditions;
 
+@NotThreadSafe
 class RFileScanner extends ScannerOptions implements Scanner {
 
   private static class RFileScannerEnvironmentImpl extends ClientServiceEnvironmentImpl {

core/src/main/java/org/apache/accumulo/core/clientImpl/ScannerOptions.java

@@ -35,6 +35,8 @@
 import java.util.TreeSet;
 import java.util.concurrent.TimeUnit;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.client.ScannerBase;
 import org.apache.accumulo.core.client.sample.SamplerConfiguration;
@@ -46,6 +48,7 @@
 import org.apache.accumulo.core.util.TextUtil;
 import org.apache.hadoop.io.Text;
 
+@NotThreadSafe
 public class ScannerOptions implements ScannerBase {
 
   protected List<IterInfo> serverSideIteratorList = Collections.emptyList();

core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchReaderIterator.java

@@ -45,6 +45,8 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.AccumuloSecurityException;
 import org.apache.accumulo.core.client.SampleNotPresentException;
@@ -355,6 +357,7 @@ private String getTableInfo() {
     return context.getPrintableTableInfoFromId(tableId);
   }
 
+  @NotThreadSafe
   private class QueryTask implements Runnable {
 
     private final String tsLocation;

core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java

@@ -41,6 +41,8 @@
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.rpc.ThriftUtil;
 import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
 import org.apache.accumulo.core.util.HostAndPort;
@@ -577,6 +579,7 @@ public TransportPoolShutdownException(String msg) {
     private static final long serialVersionUID = 1L;
   }
 
+  @NotThreadSafe
   private static class CachedTTransport extends TTransport {
 
     private final ThriftTransportKey cacheKey;

core/src/main/java/org/apache/accumulo/core/crypto/streams/BlockedInputStream.java

@@ -23,10 +23,13 @@
 import java.io.IOException;
 import java.io.InputStream;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 /**
  * Reader corresponding to BlockedOutputStream. Expects all data to be in the form of size (int)
  * data (size bytes) junk (however many bytes it takes to complete a block)
  */
+@NotThreadSafe
 public class BlockedInputStream extends InputStream {
   byte[] array;
   // ReadPos is where to start reading

core/src/main/java/org/apache/accumulo/core/fate/zookeeper/DistributedReadWriteLock.java

@@ -27,6 +27,7 @@
 import java.util.Map.Entry;
 import java.util.SortedMap;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 
@@ -111,7 +112,7 @@ static class ReadLock implements Lock {
 
     QueueLock qlock;
     byte[] userData;
-    long entry = -1;
+    AtomicLong entry;
 
     ReadLock(QueueLock qlock, byte[] userData) {
       this.qlock = qlock;
@@ -122,7 +123,7 @@ static class ReadLock implements Lock {
     ReadLock(QueueLock qlock, byte[] userData, long entry) {
       this.qlock = qlock;
       this.userData = userData;
-      this.entry = entry;
+      this.entry = new AtomicLong(entry);
     }
 
     protected LockType lockType() {
@@ -154,22 +155,27 @@ public void lockInterruptibly() throws InterruptedException {
 
     @Override
     public boolean tryLock() {
-      if (entry == -1) {
-        entry = qlock.addEntry(new ParsedLock(this.lockType(), this.userData).getLockData());
-        log.info("Added lock entry {} userData {} lockType {}", entry,
-            new String(this.userData, UTF_8), lockType());
-      }
-      SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
+      var entryVal = entry.updateAndGet(val -> {
+        if (val == -1) {
+          var newVal = qlock.addEntry(new ParsedLock(this.lockType(), this.userData).getLockData());
+          log.info("Added lock entry {} userData {} lockType {}", newVal,
+              new String(this.userData, UTF_8), lockType());
+          return newVal;
+        } else {
+          return val;
+        }
+      });
+      SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entryVal);
       for (Entry<Long,byte[]> entry : entries.entrySet()) {
         ParsedLock parsed = new ParsedLock(entry.getValue());
-        if (entry.getKey().equals(this.entry)) {
+        if (entry.getKey().equals(entryVal)) {
           return true;
         }
         if (parsed.type == LockType.WRITE) {
           return false;
         }
       }
-      throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+      throw new IllegalStateException("Did not find our own lock in the queue: " + entryVal
           + " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
     }
 
@@ -190,13 +196,14 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
 
     @Override
     public void unlock() {
-      if (entry == -1) {
-        return;
-      }
-      log.debug("Removing lock entry {} userData {} lockType {}", entry,
-          new String(this.userData, UTF_8), lockType());
-      qlock.removeEntry(entry);
-      entry = -1;
+      entry.updateAndGet(val -> {
+        if (val != -1) {
+          log.debug("Removing lock entry {} userData {} lockType {}", val,
+              new String(this.userData, UTF_8), lockType());
+          qlock.removeEntry(val);
+        }
+        return -1;
+      });
     }
 
     @Override
@@ -222,18 +229,23 @@ protected LockType lockType() {
 
     @Override
     public boolean tryLock() {
-      if (entry == -1) {
-        entry = qlock.addEntry(new ParsedLock(this.lockType(), this.userData).getLockData());
-        log.info("Added lock entry {} userData {} lockType {}", entry,
-            new String(this.userData, UTF_8), lockType());
-      }
-      SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
+      var entryVal = entry.updateAndGet(val -> {
+        if (val == -1) {
+          var newVal = qlock.addEntry(new ParsedLock(this.lockType(), this.userData).getLockData());
+          log.info("Added lock entry {} userData {} lockType {}", newVal,
+              new String(this.userData, UTF_8), lockType());
+          return newVal;
+        } else {
+          return val;
+        }
+      });
+      SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entryVal);
       Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
       if (!iterator.hasNext()) {
-        throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+        throw new IllegalStateException("Did not find our own lock in the queue: " + entryVal
             + " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
       }
-      return iterator.next().getKey().equals(entry);
+      return iterator.next().getKey().equals(entryVal);
     }
   }
 

core/src/main/java/org/apache/accumulo/core/file/BloomFilterLayer.java

@@ -35,6 +35,8 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.bloomfilter.DynamicBloomFilter;
 import org.apache.accumulo.core.classloader.ClassLoaderUtil;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
@@ -346,6 +348,7 @@ public void close() {
     }
   }
 
+  @NotThreadSafe
   public static class Reader implements FileSKVIterator {
 
     private final BloomFilterLoader bfl;

core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/SeekableByteArrayInputStream.java

@@ -23,12 +23,15 @@
 import java.io.IOException;
 import java.io.InputStream;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 
 /**
  * This class is like byte array input stream with two differences. It supports seeking and avoids
  * synchronization.
  */
+@NotThreadSafe
 public class SeekableByteArrayInputStream extends InputStream {
 
   // making this volatile for the following case

core/src/main/java/org/apache/accumulo/core/file/map/MapFileOperations.java

@@ -24,6 +24,8 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
@@ -43,6 +45,7 @@
 public class MapFileOperations extends FileOperations {
   private static final String MSG = "Map files are not supported";
 
+  @NotThreadSafe
   public static class RangeIterator implements FileSKVIterator {
 
     SortedKeyValueIterator<Key,Value> reader;

core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java

@@ -40,6 +40,8 @@
 import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.client.SampleNotPresentException;
 import org.apache.accumulo.core.client.sample.Sampler;
 import org.apache.accumulo.core.client.sample.SamplerConfiguration;
@@ -557,6 +559,7 @@ public void close() throws IOException {
     }
   }
 
+  @NotThreadSafe
   public static class Writer implements FileSKVWriter {
 
     public static final int MAX_CF_IN_DLG = 1000;
@@ -754,6 +757,7 @@ public long getLength() {
     }
   }
 
+  @NotThreadSafe
   private static class LocalityGroupReader extends LocalityGroup implements FileSKVIterator {
 
     private final CachableBlockFile.Reader reader;

core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/SimpleBufferedOutputStream.java

@@ -22,10 +22,13 @@
 import java.io.IOException;
 import java.io.OutputStream;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 /**
  * A simplified BufferedOutputStream with borrowed buffer, and allow users to see how much data have
  * been buffered.
  */
+@NotThreadSafe
 class SimpleBufferedOutputStream extends FilterOutputStream {
   protected byte[] buf; // the borrowed buffer
   protected int count = 0; // bytes used in buffer.

core/src/main/java/org/apache/accumulo/core/file/streams/BoundedRangeFileInputStream.java

@@ -21,13 +21,16 @@
 import java.io.IOException;
 import java.io.InputStream;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.hadoop.fs.Seekable;
 
 /**
  * BoundedRangeFIleInputStream abstracts a contiguous region of a Hadoop FSDataInputStream as a
  * regular input stream. One can create multiple BoundedRangeFileInputStream on top of the same
  * FSDataInputStream and they would not interfere with each other.
  */
+@NotThreadSafe
 public class BoundedRangeFileInputStream extends InputStream {
 
   private volatile boolean closed = false;

core/src/main/java/org/apache/accumulo/core/iteratorsImpl/system/ColumnFamilySkippingIterator.java

@@ -25,6 +25,8 @@
 import java.util.TreeSet;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.Column;
 import org.apache.accumulo.core.data.Key;
@@ -36,6 +38,7 @@
 import org.apache.accumulo.core.iterators.ServerSkippingIterator;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 
+@NotThreadSafe
 public class ColumnFamilySkippingIterator extends ServerSkippingIterator
     implements InterruptibleIterator {
 

core/src/main/java/org/apache/accumulo/core/iteratorsImpl/system/StatsIterator.java

@@ -26,6 +26,8 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
@@ -34,6 +36,7 @@
 import org.apache.accumulo.core.iterators.ServerWrappingIterator;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 
+@NotThreadSafe
 public class StatsIterator extends ServerWrappingIterator {
 
   private int numRead = 0;

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java

@@ -44,6 +44,8 @@
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
+import javax.annotation.concurrent.NotThreadSafe;
+
 import org.apache.accumulo.core.client.AccumuloClient;
 import org.apache.accumulo.core.client.BatchScanner;
 import org.apache.accumulo.core.client.IsolatedScanner;
@@ -89,6 +91,7 @@
  */
 public class TabletsMetadata implements Iterable<TabletMetadata>, AutoCloseable {
 
+  @NotThreadSafe
   public static class Builder implements TableRangeOptions, TableOptions, RangeOptions, Options {
 
     private final List<Text> families = new ArrayList<>();

core/src/main/java/org/apache/accumulo/core/singletons/SingletonManager.java

@@ -27,6 +27,8 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
 /**
  * This class automates management of static singletons that maintain state for Accumulo clients.
  * Historically, Accumulo client code that used Connector had no control over these singletons. The
@@ -81,6 +83,8 @@ public enum Mode {
   private static List<SingletonService> services;
 
   @VisibleForTesting
+  @SuppressFBWarnings(value = "AT_NONATOMIC_64BIT_PRIMITIVE",
+      justification = "only called in static init block and testing, no sync needed")
   static void reset() {
     reservations = 0;
     mode = Mode.CLIENT;