Address ErrorProne warnings

core/src/main/java/com/datastax/oss/driver/api/core/CoreProtocolVersion.java

@@ -47,6 +47,7 @@ public enum CoreProtocolVersion implements ProtocolVersion {
     this.beta = beta;
   }
 
+  @Override
   public int getCode() {
     return code;
   }

core/src/main/java/com/datastax/oss/driver/api/core/config/DriverConfigLoader.java

@@ -35,5 +35,6 @@ public interface DriverConfigLoader extends AutoCloseable {
    * Called when the cluster closes. This is a good time to release any external resource, for
    * example cancel a scheduled reloading task.
    */
+  @Override
   void close();
 }

core/src/main/java/com/datastax/oss/driver/api/core/connection/ExponentialReconnectionPolicy.java

@@ -58,7 +58,7 @@ public ExponentialReconnectionPolicy(DriverContext context) {
 
     // Maximum number of attempts after which we overflow
     int ceil = (baseDelayMs & (baseDelayMs - 1)) == 0 ? 0 : 1;
-    this.maxAttempts = 64 - Long.numberOfLeadingZeros(Long.MAX_VALUE / baseDelayMs) - ceil;
+    this.maxAttempts = 64L - Long.numberOfLeadingZeros(Long.MAX_VALUE / baseDelayMs) - ceil;
   }
 
   /**

core/src/main/java/com/datastax/oss/driver/api/core/cql/SimpleStatementBuilder.java

@@ -121,6 +121,7 @@ public SimpleStatementBuilder clearNamedValues() {
     return this;
   }
 
+  @Override
   public SimpleStatement build() {
     return new DefaultSimpleStatement(
         query,

core/src/main/java/com/datastax/oss/driver/api/core/data/CqlDuration.java

@@ -351,7 +351,10 @@ public Builder(boolean isNegative) {
     public Builder addYears(long numberOfYears) {
       validateOrder(1);
       validateMonths(numberOfYears, MONTHS_PER_YEAR);
-      months += numberOfYears * MONTHS_PER_YEAR;
+      // Cast to avoid http://errorprone.info/bugpattern/NarrowingCompoundAssignment
+      // We could also change the method to accept an int, but keeping long allows us to keep the
+      // calling code generic.
+      months += (int) numberOfYears * MONTHS_PER_YEAR;
       return this;
     }
 
@@ -364,7 +367,7 @@ public Builder addYears(long numberOfYears) {
     public Builder addMonths(long numberOfMonths) {
       validateOrder(2);
       validateMonths(numberOfMonths, 1);
-      months += numberOfMonths;
+      months += (int) numberOfMonths;
       return this;
     }
 
@@ -377,7 +380,7 @@ public Builder addMonths(long numberOfMonths) {
     public Builder addWeeks(long numberOfWeeks) {
       validateOrder(3);
       validateDays(numberOfWeeks, DAYS_PER_WEEK);
-      days += numberOfWeeks * DAYS_PER_WEEK;
+      days += (int) numberOfWeeks * DAYS_PER_WEEK;
       return this;
     }
 
@@ -390,7 +393,7 @@ public Builder addWeeks(long numberOfWeeks) {
     public Builder addDays(long numberOfDays) {
       validateOrder(4);
       validateDays(numberOfDays, 1);
-      days += numberOfDays;
+      days += (int) numberOfDays;
       return this;
     }
 

core/src/main/java/com/datastax/oss/driver/api/core/type/CustomType.java

@@ -29,6 +29,7 @@ default String asCql(boolean includeFrozen, boolean pretty) {
     return String.format("'%s'", getClassName());
   }
 
+  @Override
   default int getProtocolCode() {
     return ProtocolConstants.DataType.CUSTOM;
   }

core/src/main/java/com/datastax/oss/driver/api/core/type/ListType.java

@@ -28,6 +28,7 @@ default String asCql(boolean includeFrozen, boolean pretty) {
     return String.format(template, getElementType().asCql(includeFrozen, pretty));
   }
 
+  @Override
   default int getProtocolCode() {
     return ProtocolConstants.DataType.LIST;
   }

core/src/main/java/com/datastax/oss/driver/api/core/type/MapType.java

@@ -33,6 +33,7 @@ default String asCql(boolean includeFrozen, boolean pretty) {
         getValueType().asCql(includeFrozen, pretty));
   }
 
+  @Override
   default int getProtocolCode() {
     return ProtocolConstants.DataType.MAP;
   }

core/src/main/java/com/datastax/oss/driver/api/core/type/SetType.java

@@ -28,6 +28,7 @@ default String asCql(boolean includeFrozen, boolean pretty) {
     return String.format(template, getElementType().asCql(includeFrozen, pretty));
   }
 
+  @Override
   default int getProtocolCode() {
     return ProtocolConstants.DataType.SET;
   }

core/src/main/java/com/datastax/oss/driver/api/core/type/TupleType.java

@@ -47,6 +47,7 @@ default String asCql(boolean includeFrozen, boolean pretty) {
     return builder.toString();
   }
 
+  @Override
   default int getProtocolCode() {
     return ProtocolConstants.DataType.TUPLE;
   }

core/src/main/java/com/datastax/oss/driver/api/core/type/UserDefinedType.java

@@ -91,6 +91,7 @@ default String describeWithChildren(boolean pretty) {
     return describe(pretty);
   }
 
+  @Override
   default int getProtocolCode() {
     return ProtocolConstants.DataType.UDT;
   }

core/src/main/java/com/datastax/oss/driver/internal/core/channel/ChannelHandlerRequest.java

@@ -17,6 +17,7 @@
 
 import com.datastax.oss.driver.api.core.DriverTimeoutException;
 import com.datastax.oss.driver.internal.core.util.ProtocolUtils;
+import com.datastax.oss.driver.internal.core.util.concurrent.UncaughtExceptions;
 import com.datastax.oss.protocol.internal.Frame;
 import com.datastax.oss.protocol.internal.Message;
 import com.datastax.oss.protocol.internal.response.Error;
@@ -91,7 +92,7 @@ private void onTimeout() {
     fail(new DriverTimeoutException(describe() + ": timed out after " + timeoutMillis + " ms"));
     if (!channel.closeFuture().isDone()) {
       // Cancel the response callback
-      channel.writeAndFlush(this);
+      channel.writeAndFlush(this).addListener(UncaughtExceptions::log);
     }
   }
 

core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannel.java

@@ -17,6 +17,7 @@
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.ProtocolVersion;
+import com.datastax.oss.driver.internal.core.util.concurrent.UncaughtExceptions;
 import com.datastax.oss.protocol.internal.Message;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
@@ -82,7 +83,7 @@ public Future<Void> write(
   public void cancel(ResponseCallback responseCallback) {
     // To avoid creating an extra message, we adopt the convention that writing the callback
     // directly means cancellation
-    writeCoalescer.writeAndFlush(channel, responseCallback);
+    writeCoalescer.writeAndFlush(channel, responseCallback).addListener(UncaughtExceptions::log);
   }
 
   /**
@@ -149,10 +150,12 @@ public SocketAddress localAddress() {
    * be allowed to complete before the underlying channel is closed.
    */
   public Future<Void> close() {
-    if (closing.compareAndSet(false, true)) {
+    if (closing.compareAndSet(false, true) && channel.isOpen()) {
       // go through the coalescer: this guarantees that we won't reject writes that were submitted
       // before, but had not been coalesced yet.
-      writeCoalescer.writeAndFlush(channel, GRACEFUL_CLOSE_MESSAGE);
+      writeCoalescer
+          .writeAndFlush(channel, GRACEFUL_CLOSE_MESSAGE)
+          .addListener(UncaughtExceptions::log);
     }
     return channel.closeFuture();
   }
@@ -163,8 +166,10 @@ public Future<Void> close() {
    */
   public Future<Void> forceClose() {
     this.close();
-    if (forceClosing.compareAndSet(false, true)) {
-      writeCoalescer.writeAndFlush(channel, FORCEFUL_CLOSE_MESSAGE);
+    if (forceClosing.compareAndSet(false, true) && channel.isOpen()) {
+      writeCoalescer
+          .writeAndFlush(channel, FORCEFUL_CLOSE_MESSAGE)
+          .addListener(UncaughtExceptions::log);
     }
     return channel.closeFuture();
   }

core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultTraceEvent.java

@@ -56,6 +56,7 @@ public InetAddress getSource() {
     return source;
   }
 
+  @Override
   public int getSourceElapsedMicros() {
     return sourceElapsedMicros;
   }

core/src/main/java/com/datastax/oss/driver/internal/core/cql/MultiPageResultSet.java

@@ -24,9 +24,10 @@
 import com.datastax.oss.driver.internal.core.util.CountingIterator;
 import com.datastax.oss.driver.internal.core.util.concurrent.BlockingOperation;
 import com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures;
+import java.util.ArrayDeque;
 import java.util.ArrayList;
+import java.util.Deque;
 import java.util.Iterator;
-import java.util.LinkedList;
 import java.util.List;
 
 public class MultiPageResultSet implements ResultSet {
@@ -81,7 +82,7 @@ public boolean wasApplied() {
 
   private class RowIterator extends CountingIterator<Row> {
     // The pages fetched so far. The first is the one we're currently iterating.
-    private LinkedList<AsyncResultSet> pages = new LinkedList<>();
+    private Deque<AsyncResultSet> pages = new ArrayDeque<>();
     private Iterator<Row> currentRows;
 
     private RowIterator(AsyncResultSet firstPage) {

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultNode.java

@@ -98,6 +98,7 @@ public NodeState getState() {
     return state;
   }
 
+  @Override
   public int getOpenConnections() {
     return openConnections;
   }

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/DefaultIndexMetadata.java

@@ -70,6 +70,7 @@ public String getTarget() {
     return target;
   }
 
+  @Override
   public Map<String, String> getOptions() {
     return options;
   }

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/parsing/RelationParser.java

@@ -114,7 +114,7 @@ public static void appendOptions(Map<CqlIdentifier, Object> options, ScriptBuild
    * The columns of the system table that are turned into entries in {@link
    * RelationMetadata#getOptions()}.
    */
-  public static final Map<String, TypeCodec<?>> OPTION_CODECS =
+  public static final ImmutableMap<String, TypeCodec<?>> OPTION_CODECS =
       ImmutableMap.<String, TypeCodec<?>>builder()
           .put("bloom_filter_fp_chance", TypeCodecs.DOUBLE)
           // In C* <= 2.2, this is a string, not a map (this is special-cased in parseOptions):

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/parsing/TableParser.java

@@ -292,9 +292,9 @@ private static String buildLegacyIndexTarget(ColumnMetadata column, Map<String,
     if (options.containsKey("index_keys_and_values")) {
       return String.format("entries(%s)", columnName);
     }
-    if (columnType instanceof ListType && ((ListType) columnType).isFrozen()
-        || columnType instanceof SetType && ((SetType) columnType).isFrozen()
-        || columnType instanceof MapType && ((MapType) columnType).isFrozen()) {
+    if ((columnType instanceof ListType && ((ListType) columnType).isFrozen())
+        || (columnType instanceof SetType && ((SetType) columnType).isFrozen())
+        || (columnType instanceof MapType && ((MapType) columnType).isFrozen())) {
       return String.format("full(%s)", columnName);
     }
     // Note: the keyword 'values' is not accepted as a valid index target function until 3.0

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/DefaultSchemaQueriesFactory.java

@@ -35,6 +35,7 @@ public DefaultSchemaQueriesFactory(InternalDriverContext context) {
     this.context = context;
   }
 
+  @Override
   public SchemaQueries newInstance(CompletableFuture<Metadata> refreshFuture) {
     String logPrefix = context.sessionName();
 

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

@@ -33,9 +33,9 @@
 import com.google.common.annotations.VisibleForTesting;
 import java.nio.ByteBuffer;
 import java.time.Duration;
+import java.util.ArrayDeque;
 import java.util.Collections;
 import java.util.HashSet;
-import java.util.LinkedList;
 import java.util.Map;
 import java.util.Queue;
 import java.util.Set;
@@ -161,7 +161,7 @@ private void gatherServerIds(AdminResult rows, Throwable error) {
 
   private void gatherPayloadsToReprepare() {
     assert channel.eventLoop().inEventLoop();
-    toReprepare = new LinkedList<>();
+    toReprepare = new ArrayDeque<>();
     for (RepreparePayload payload : repreparePayloads.values()) {
       if (serverKnownIds.contains(payload.id)) {
         LOG.trace(

core/src/main/java/com/datastax/oss/driver/internal/core/type/util/VIntCoding.java

@@ -96,9 +96,9 @@ private static int firstByteValueMask(int extraBytesToRead) {
     return 0xff >> extraBytesToRead;
   }
 
-  private static int encodeExtraBytesToRead(int extraBytesToRead) {
+  private static byte encodeExtraBytesToRead(int extraBytesToRead) {
     // because we have an extra bit in the value mask, we just need to invert it
-    return ~firstByteValueMask(extraBytesToRead);
+    return (byte) ~firstByteValueMask(extraBytesToRead);
   }
 
   private static int numberOfExtraBytesToRead(int firstByte) {

core/src/main/java/com/datastax/oss/driver/internal/core/util/DirectedGraph.java

@@ -21,9 +21,9 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multimap;
+import java.util.ArrayDeque;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
@@ -67,7 +67,7 @@ public List<V> topologicalSort() {
     Preconditions.checkState(!wasSorted);
     wasSorted = true;
 
-    Queue<V> queue = new LinkedList<V>();
+    Queue<V> queue = new ArrayDeque<>();
 
     for (Map.Entry<V, Integer> entry : vertices.entrySet()) {
       if (entry.getValue() == 0) {

core/src/main/java/com/datastax/oss/driver/internal/core/util/Strings.java

@@ -255,7 +255,7 @@ public static boolean isLongLiteral(String str) {
 
   private Strings() {}
 
-  private static final Set<String> RESERVED_KEYWORDS =
+  private static final ImmutableSet<String> RESERVED_KEYWORDS =
       ImmutableSet.of(
           // See https://github.com/apache/cassandra/blob/trunk/doc/cql3/CQL.textile#appendixA
           "add",

core/src/test/java/com/datastax/oss/driver/api/core/CassandraVersionAssert.java

@@ -59,6 +59,7 @@ public CassandraVersionAssert hasNextStable(String version) {
     return this;
   }
 
+  @Override
   public CassandraVersionAssert hasToString(String string) {
     Assertions.assertThat(actual.toString()).isEqualTo(string);
     return this;

core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryAvailableIdsTest.java

@@ -36,6 +36,7 @@ public class ChannelFactoryAvailableIdsTest extends ChannelFactoryTestBase {
   @Mock private ResponseCallback responseCallback;
 
   @Before
+  @Override
   public void setup() throws InterruptedException {
     super.setup();
     Mockito.when(defaultConfigProfile.isDefined(CoreDriverOption.PROTOCOL_VERSION))

core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryTestBase.java

@@ -121,7 +121,7 @@ public void setup() throws InterruptedException {
     Mockito.when(defaultConfigProfile.getInt(CoreDriverOption.CONNECTION_MAX_REQUESTS))
         .thenReturn(1);
     Mockito.when(defaultConfigProfile.getDuration(CoreDriverOption.CONNECTION_HEARTBEAT_INTERVAL))
-        .thenReturn(Duration.ofMillis(30000));
+        .thenReturn(Duration.ofSeconds(30));
 
     Mockito.when(context.protocolVersionRegistry()).thenReturn(protocolVersionRegistry);
     Mockito.when(context.nettyOptions()).thenReturn(nettyOptions);

core/src/test/java/com/datastax/oss/driver/internal/core/channel/DriverChannelTest.java

@@ -25,7 +25,7 @@
 import io.netty.channel.ChannelPromise;
 import io.netty.util.concurrent.Future;
 import java.util.AbstractMap;
-import java.util.LinkedList;
+import java.util.ArrayDeque;
 import java.util.Map;
 import java.util.Queue;
 import org.junit.Before;
@@ -142,7 +142,7 @@ public void should_wait_for_coalesced_writes_when_closing_forcefully() {
   // Simple implementation that holds all the writes, and flushes them when it's explicitly
   // triggered.
   private class MockWriteCoalescer implements WriteCoalescer {
-    private Queue<Map.Entry<Object, ChannelPromise>> messages = new LinkedList<>();
+    private Queue<Map.Entry<Object, ChannelPromise>> messages = new ArrayDeque<>();
 
     @Override
     public ChannelFuture writeAndFlush(Channel channel, Object message) {

core/src/test/java/com/datastax/oss/driver/internal/core/channel/MockChannelFactoryHelper.java

@@ -20,8 +20,9 @@
 import com.google.common.collect.MultimapBuilder;
 import com.google.common.collect.Sets;
 import java.net.SocketAddress;
+import java.util.ArrayDeque;
+import java.util.Deque;
 import java.util.HashMap;
-import java.util.LinkedList;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -148,7 +149,7 @@ public MockChannelFactoryHelper build() {
 
     private void stub() {
       for (SocketAddress address : invocations.keySet()) {
-        LinkedList<CompletionStage<DriverChannel>> results = new LinkedList<>();
+        Deque<CompletionStage<DriverChannel>> results = new ArrayDeque<>();
         for (Object object : invocations.get(address)) {
           if (object instanceof DriverChannel) {
             results.add(CompletableFuture.completedFuture(((DriverChannel) object)));

core/src/test/java/com/datastax/oss/driver/internal/core/channel/MockResponseCallback.java

@@ -16,12 +16,12 @@
 package com.datastax.oss.driver.internal.core.channel;
 
 import com.datastax.oss.protocol.internal.Frame;
-import java.util.LinkedList;
+import java.util.ArrayDeque;
 import java.util.Queue;
 
 class MockResponseCallback implements ResponseCallback {
   private final boolean holdStreamId;
-  private final Queue<Object> responses = new LinkedList<>();
+  private final Queue<Object> responses = new ArrayDeque<>();
 
   volatile int streamId = -1;