From f8fc0311b65b3d82737352f3d01483c0334a6867 Mon Sep 17 00:00:00 2001 From: Ariel Weisberg Date: Fri, 27 Nov 2015 11:40:16 +0100 Subject: [PATCH] fix 2.2 eclipse-warnings patch by Ariel Weisberg; reviewed by Robert Stupp for CASSANDRA-9800 --- .../cassandra/cache/AutoSavingCache.java | 1 + .../db/WindowsFailedSnapshotTracker.java | 43 ++--- .../db/commitlog/CommitLogReplayer.java | 3 +- .../AbstractColumnFamilyInputFormat.java | 4 +- .../hadoop/cql3/CqlRecordWriter.java | 168 ++++++++++-------- .../hadoop/pig/CqlNativeStorage.java | 6 +- .../util/ChecksummedRandomAccessReader.java | 29 ++- .../cassandra/io/util/SegmentedFile.java | 1 + .../cassandra/net/IncomingTcpConnection.java | 3 +- 9 files changed, 154 insertions(+), 104 deletions(-) diff --git a/src/java/org/apache/cassandra/cache/AutoSavingCache.java b/src/java/org/apache/cassandra/cache/AutoSavingCache.java index c08925d735..2c6820ef2a 100644 --- a/src/java/org/apache/cassandra/cache/AutoSavingCache.java +++ b/src/java/org/apache/cassandra/cache/AutoSavingCache.java @@ -318,6 +318,7 @@ public class AutoSavingCache extends InstrumentingCache filter(Mutation mutation); @@ -273,6 +273,7 @@ public class CommitLogReplayer } } + @SuppressWarnings("resource") public void recover(File file, boolean tolerateTruncation) throws IOException { CommitLogDescriptor desc = CommitLogDescriptor.fromFileName(file.getName()); diff --git a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java index 148c08a03b..3c088c2ffa 100644 --- a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java +++ b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java @@ -117,9 +117,9 @@ public abstract class AbstractColumnFamilyInputFormat extends InputFormat< } } - try (Cluster cluster = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf)) + try (Cluster cluster = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf); + Session session = cluster.connect()) { - Session session = cluster.connect(); Metadata metadata = session.getCluster().getMetadata(); for (TokenRange range : masterRangeNodes.keySet()) diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java index 14e24fb4b1..84102a5eb9 100644 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java +++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java @@ -43,7 +43,7 @@ import org.apache.hadoop.util.Progressable; /** * The CqlRecordWriter maps the output <key, value> * pairs to a Cassandra table. In particular, it applies the binded variables - * in the value to the prepared statement, which it associates with the key, and in + * in the value to the prepared statement, which it associates with the key, and in * turn the responsible endpoint. * *

@@ -112,11 +112,11 @@ class CqlRecordWriter extends RecordWriter, List(); + String keyspace = ConfigHelper.getOutputKeyspace(conf); - try (Cluster cluster = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf)) + try (Cluster cluster = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf); + Session client = cluster.connect(keyspace)) { - String keyspace = ConfigHelper.getOutputKeyspace(conf); - Session client = cluster.connect(keyspace); ringCache = new NativeRingCache(conf); if (client != null) { @@ -179,7 +179,7 @@ class CqlRecordWriter extends RecordWriter, List, List, List bindVariables; - try + outer: + while (run || !queue.isEmpty()) { - bindVariables = queue.take(); - } - catch (InterruptedException e) - { - // re-check loop condition after interrupt - continue; - } - - ListIterator iter = endpoints.listIterator(); - while (true) - { - // send the mutation to the last-used endpoint. first time through, this will NPE harmlessly. - if (session != null) + List bindVariables; + try { + bindVariables = queue.take(); + } + catch (InterruptedException e) + { + // re-check loop condition after interrupt + continue; + } + + ListIterator iter = endpoints.listIterator(); + while (true) + { + // send the mutation to the last-used endpoint. first time through, this will NPE harmlessly. + if (session != null) + { + try + { + int i = 0; + PreparedStatement statement = preparedStatement(session); + while (bindVariables != null) + { + BoundStatement boundStatement = new BoundStatement(statement); + for (int columnPosition = 0; columnPosition < bindVariables.size(); columnPosition++) + { + boundStatement.setBytesUnsafe(columnPosition, bindVariables.get(columnPosition)); + } + session.execute(boundStatement); + i++; + + if (i >= batchThreshold) + break; + bindVariables = queue.poll(); + } + break; + } + catch (Exception e) + { + closeInternal(); + if (!iter.hasNext()) + { + lastException = new IOException(e); + break outer; + } + } + } + + // attempt to connect to a different endpoint try { - int i = 0; - PreparedStatement statement = preparedStatement(session); - while (bindVariables != null) - { - BoundStatement boundStatement = new BoundStatement(statement); - for (int columnPosition = 0; columnPosition < bindVariables.size(); columnPosition++) - { - boundStatement.setBytesUnsafe(columnPosition, bindVariables.get(columnPosition)); - } - session.execute(boundStatement); - i++; - - if (i >= batchThreshold) - break; - bindVariables = queue.poll(); - } - break; + InetAddress address = iter.next(); + String host = address.getHostName(); + cluster = CqlConfigHelper.getOutputCluster(host, conf); + closeSession(session); + session = cluster.connect(); } catch (Exception e) { + //If connection died due to Interrupt, just try connecting to the endpoint again. + //There are too many ways for the Thread.interrupted() state to be cleared, so + //we can't rely on that here. Until the java driver gives us a better way of knowing + //that this exception came from an InterruptedException, this is the best solution. + if (canRetryDriverConnection(e)) + { + iter.previous(); + } closeInternal(); - if (!iter.hasNext()) + + // Most exceptions mean something unexpected went wrong to that endpoint, so + // we should try again to another. Other exceptions (auth or invalid request) are fatal. + if ((e instanceof AuthenticationException || e instanceof InvalidQueryException) || !iter.hasNext()) { lastException = new IOException(e); break outer; } - } - } - - // attempt to connect to a different endpoint - try - { - InetAddress address = iter.next(); - String host = address.getHostName(); - cluster = CqlConfigHelper.getOutputCluster(host, conf); - session = cluster.connect(); - } - catch (Exception e) - { - //If connection died due to Interrupt, just try connecting to the endpoint again. - //There are too many ways for the Thread.interrupted() state to be cleared, so - //we can't rely on that here. Until the java driver gives us a better way of knowing - //that this exception came from an InterruptedException, this is the best solution. - if (canRetryDriverConnection(e)) - { - iter.previous(); - } - closeInternal(); - - // Most exceptions mean something unexpected went wrong to that endpoint, so - // we should try again to another. Other exceptions (auth or invalid request) are fatal. - if ((e instanceof AuthenticationException || e instanceof InvalidQueryException) || !iter.hasNext()) - { - lastException = new IOException(e); - break outer; } } } } + finally + { + closeSession(session); + } + // close all our connections once we are done. closeInternal(); } @@ -489,9 +513,9 @@ class CqlRecordWriter extends RecordWriter, List(); metadata = session.getCluster().getMetadata(); Set ranges = metadata.getTokenRanges(); diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java index 74058b1de2..8831cf2ee1 100644 --- a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java +++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java @@ -690,7 +690,7 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo else throw new IOException("bulk_insert_statement is missing in input url parameter"); if (bulkTableAlias != null) - CqlBulkOutputFormat.setTableAlias(conf, bulkTableAlias, column_family); + CqlBulkOutputFormat.setTableAlias(conf, bulkTableAlias, column_family); CqlBulkOutputFormat.setDeleteSourceOnSuccess(conf, bulkDeleteSourceOnSuccess); if (bulkOutputLocation != null) conf.set(CqlBulkRecordWriter.OUTPUT_LOCATION, bulkOutputLocation); @@ -724,9 +724,9 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo // Only get the schema if we haven't already gotten it if (!properties.containsKey(signature)) { - try (Cluster cluster = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf), conf)) + try (Cluster cluster = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf), conf); + Session client = cluster.connect()) { - Session client = cluster.connect(); client.execute("USE " + keyspace); // compose the CfDef for the columfamily diff --git a/src/java/org/apache/cassandra/io/util/ChecksummedRandomAccessReader.java b/src/java/org/apache/cassandra/io/util/ChecksummedRandomAccessReader.java index 442236de35..9015b619a9 100644 --- a/src/java/org/apache/cassandra/io/util/ChecksummedRandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/ChecksummedRandomAccessReader.java @@ -48,15 +48,36 @@ public class ChecksummedRandomAccessReader extends RandomAccessReader this.file = file; } + @SuppressWarnings("resource") public static ChecksummedRandomAccessReader open(File file, File crcFile) throws IOException { try (ChannelProxy channel = new ChannelProxy(file)) { RandomAccessReader crcReader = RandomAccessReader.open(crcFile); - @SuppressWarnings("resource") - DataIntegrityMetadata.ChecksumValidator validator = - new DataIntegrityMetadata.ChecksumValidator(new Adler32(), crcReader, file.getPath()); - return new ChecksummedRandomAccessReader(file, channel, validator); + boolean closeCrcReader = true; + try + { + DataIntegrityMetadata.ChecksumValidator validator = + new DataIntegrityMetadata.ChecksumValidator(new Adler32(), crcReader, file.getPath()); + closeCrcReader = false; + boolean closeValidator = true; + try + { + ChecksummedRandomAccessReader retval = new ChecksummedRandomAccessReader(file, channel, validator); + closeValidator = false; + return retval; + } + finally + { + if (closeValidator) + validator.close(); + } + } + finally + { + if (closeCrcReader) + crcReader.close(); + } } } diff --git a/src/java/org/apache/cassandra/io/util/SegmentedFile.java b/src/java/org/apache/cassandra/io/util/SegmentedFile.java index 30707d82ae..553cc0d2b4 100644 --- a/src/java/org/apache/cassandra/io/util/SegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/SegmentedFile.java @@ -179,6 +179,7 @@ public abstract class SegmentedFile extends SharedCloseableImpl return complete(path, -1L); } + @SuppressWarnings("resource") public SegmentedFile complete(String path, long overrideLength) { ChannelProxy channelCopy = getChannel(path); diff --git a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java index f6652b0767..a9721142f9 100644 --- a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java +++ b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java @@ -108,7 +108,7 @@ public class IncomingTcpConnection extends Thread implements Closeable close(); } } - + @Override public void close() { @@ -164,6 +164,7 @@ public class IncomingTcpConnection extends Thread implements Closeable } else { + @SuppressWarnings("resource") ReadableByteChannel channel = socket.getChannel(); in = new NIODataInputStream(channel != null ? channel : Channels.newChannel(socket.getInputStream()), BUFFER_SIZE); }