Switch AtomicBoolean to volatile for closed flag

Author: jasonk000

core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottler.java

@@ -28,7 +28,6 @@
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.util.Deque;
 import java.util.concurrent.ConcurrentLinkedDeque;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import net.jcip.annotations.ThreadSafe;
 import org.slf4j.Logger;
@@ -64,7 +63,7 @@ public class ConcurrencyLimitingRequestThrottler implements RequestThrottler {
   private final AtomicInteger concurrentRequests = new AtomicInteger(0);
   private final Deque<Throttled> queue = new ConcurrentLinkedDeque<>();
   private final AtomicInteger queueSize = new AtomicInteger(0);
-  private final AtomicBoolean closed = new AtomicBoolean(false);
+  private volatile boolean closed = false;
 
   public ConcurrencyLimitingRequestThrottler(DriverContext context) {
     this.logPrefix = context.getSessionName();
@@ -81,7 +80,7 @@ public ConcurrencyLimitingRequestThrottler(DriverContext context) {
 
   @Override
   public void register(@NonNull Throttled request) {
-    if (closed.get()) {
+    if (closed) {
       LOG.trace("[{}] Rejecting request after shutdown", logPrefix);
       fail(request, "The session is shutting down");
       return;
@@ -114,7 +113,7 @@ public void register(@NonNull Throttled request) {
       // Double-check that we were still supposed to be enqueued; it is possible
       // that the session was closed while we were enqueuing, it's also possible
       // that it is right now removing the request, so we need to check both
-      if (closed.get()) {
+      if (closed) {
         if (queue.remove(request)) {
           queueSize.decrementAndGet();
           LOG.trace("[{}] Rejecting late request after shutdown", logPrefix);
@@ -148,7 +147,7 @@ public void signalError(@NonNull Throttled request, @NonNull Throwable error) {
   @Override
   public void signalTimeout(@NonNull Throttled request) {
     Throttled nextRequest = null;
-    if (!closed.get()) {
+    if (!closed) {
       if (queue.remove(request)) { // The request timed out before it was active
         queueSize.decrementAndGet();
         LOG.trace("[{}] Removing timed out request from the queue", logPrefix);
@@ -165,7 +164,7 @@ public void signalTimeout(@NonNull Throttled request) {
   @Override
   public void signalCancel(@NonNull Throttled request) {
     Throttled nextRequest = null;
-    if (!closed.get()) {
+    if (!closed) {
       if (queue.remove(request)) { // The request has been cancelled before it was active
         queueSize.decrementAndGet();
         LOG.trace("[{}] Removing cancelled request from the queue", logPrefix);
@@ -181,7 +180,7 @@ public void signalCancel(@NonNull Throttled request) {
 
   @Nullable
   private Throttled onRequestDoneAndDequeNext() {
-    if (!closed.get()) {
+    if (!closed) {
       Throttled nextRequest = queue.poll();
       if (nextRequest == null) {
         concurrentRequests.decrementAndGet();
@@ -198,7 +197,7 @@ private Throttled onRequestDoneAndDequeNext() {
 
   @Override
   public void close() {
-    closed.set(true);
+    closed = true;
 
     LOG.debug("[{}] Rejecting {} queued requests after shutdown", logPrefix, queueSize.get());
     Throttled request;