diff --git a/CHANGES.txt b/CHANGES.txt index a5b32205ba..ad2845f459 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.2 + * Static Analysis to warn on unsafe use of Autocloseable instances (CASSANDRA-9431) * Update commitlog archiving examples now that commitlog segments are not recycled (CASSANDRA-9350) * Extend Transactional API to sstable lifecycle management (CASSANDRA-8568) diff --git a/build.xml b/build.xml index cbedf201d7..bb401eed2b 100644 --- a/build.xml +++ b/build.xml @@ -68,8 +68,8 @@ - - + + @@ -114,6 +114,8 @@ + + @@ -1047,7 +1049,7 @@ - @@ -1406,7 +1408,7 @@ @@ -1894,6 +1896,40 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + extends InstrumentingCache implements ICache map.clear(); } + @SuppressWarnings("resource") public V get(K key) { RefCountedMemory mem = map.get(key); @@ -172,6 +173,7 @@ public class SerializingCache implements ICache } } + @SuppressWarnings("resource") public void put(K key, V value) { RefCountedMemory mem = serialize(value); @@ -193,6 +195,7 @@ public class SerializingCache implements ICache old.unreference(); } + @SuppressWarnings("resource") public boolean putIfAbsent(K key, V value) { RefCountedMemory mem = serialize(value); @@ -216,6 +219,7 @@ public class SerializingCache implements ICache return old == null; } + @SuppressWarnings("resource") public boolean replace(K key, V oldToReplace, V value) { // if there is no old value in our map, we fail @@ -259,6 +263,7 @@ public class SerializingCache implements ICache public void remove(K key) { + @SuppressWarnings("resource") RefCountedMemory mem = map.remove(key); if (mem != null) mem.unreference(); diff --git a/src/java/org/apache/cassandra/db/BatchlogManager.java b/src/java/org/apache/cassandra/db/BatchlogManager.java index f5137fdfe0..dd84ac88b9 100644 --- a/src/java/org/apache/cassandra/db/BatchlogManager.java +++ b/src/java/org/apache/cassandra/db/BatchlogManager.java @@ -148,20 +148,17 @@ public class BatchlogManager implements BatchlogManagerMBean private static ByteBuffer serializeMutations(Collection mutations, int version) { - DataOutputBuffer buf = new DataOutputBuffer(); - - try + try (DataOutputBuffer buf = new DataOutputBuffer()) { buf.writeInt(mutations.size()); for (Mutation mutation : mutations) Mutation.serializer.serialize(mutation, buf, version); + return buf.buffer(); } catch (IOException e) { throw new AssertionError(); // cannot happen. } - - return buf.buffer(); } private void replayAllFailedBatches() throws ExecutionException, InterruptedException diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java index 006ced78c4..a7243a2fba 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamily.java +++ b/src/java/org/apache/cassandra/db/ColumnFamily.java @@ -521,9 +521,11 @@ public abstract class ColumnFamily implements Iterable, IRowCacheEntry public ByteBuffer toBytes() { - DataOutputBuffer out = new DataOutputBuffer(); - serializer.serialize(this, out, MessagingService.current_version); - return ByteBuffer.wrap(out.getData(), 0, out.getLength()); + try (DataOutputBuffer out = new DataOutputBuffer()) + { + serializer.serialize(this, out, MessagingService.current_version); + return ByteBuffer.wrap(out.getData(), 0, out.getLength()); + } } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index b41ac75b99..63ffb167cc 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1795,6 +1795,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean return repairedSSTables; } + @SuppressWarnings("resource") public RefViewFragment selectAndReference(Function> filter) { while (true) @@ -1966,6 +1967,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean * * @param range The range of keys and columns within those keys to fetch */ + @SuppressWarnings("resource") private AbstractScanIterator getSequentialIterator(final DataRange range, long now) { assert !(range.keyRange() instanceof Range) || !((Range)range.keyRange()).isWrapAround() || range.keyRange().right.isMinimum() : range.keyRange(); @@ -1980,24 +1982,27 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { protected Row computeNext() { - // pull a row out of the iterator - if (!iterator.hasNext()) - return endOfData(); + while (true) + { + // pull a row out of the iterator + if (!iterator.hasNext()) + return endOfData(); - Row current = iterator.next(); - DecoratedKey key = current.key; + Row current = iterator.next(); + DecoratedKey key = current.key; - if (!range.stopKey().isMinimum() && range.stopKey().compareTo(key) < 0) - return endOfData(); + if (!range.stopKey().isMinimum() && range.stopKey().compareTo(key) < 0) + return endOfData(); - // skipping outside of assigned range - if (!range.contains(key)) - return computeNext(); + // skipping outside of assigned range + if (!range.contains(key)) + continue; - if (logger.isTraceEnabled()) - logger.trace("scanned {}", metadata.getKeyValidator().getString(key.getKey())); + if (logger.isTraceEnabled()) + logger.trace("scanned {}", metadata.getKeyValidator().getString(key.getKey())); - return current; + return current; + } } public void close() throws IOException diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index 55b0bfe9a4..ccf92be84d 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -423,19 +423,21 @@ public class Memtable implements Comparable private static int estimateRowOverhead(final int count) { // calculate row overhead - final OpOrder.Group group = new OpOrder().start(); - int rowOverhead; - MemtableAllocator allocator = MEMORY_POOL.newAllocator(); - ConcurrentNavigableMap rows = new ConcurrentSkipListMap<>(); - final Object val = new Object(); - for (int i = 0 ; i < count ; i++) - rows.put(allocator.clone(new BufferDecoratedKey(new LongToken(i), ByteBufferUtil.EMPTY_BYTE_BUFFER), group), val); - double avgSize = ObjectSizes.measureDeep(rows) / (double) count; - rowOverhead = (int) ((avgSize - Math.floor(avgSize)) < 0.05 ? Math.floor(avgSize) : Math.ceil(avgSize)); - rowOverhead -= ObjectSizes.measureDeep(new LongToken(0)); - rowOverhead += AtomicBTreeColumns.EMPTY_SIZE; - allocator.setDiscarding(); - allocator.setDiscarded(); - return rowOverhead; + try (final OpOrder.Group group = new OpOrder().start()) + { + int rowOverhead; + MemtableAllocator allocator = MEMORY_POOL.newAllocator(); + ConcurrentNavigableMap rows = new ConcurrentSkipListMap<>(); + final Object val = new Object(); + for (int i = 0; i < count; i++) + rows.put(allocator.clone(new BufferDecoratedKey(new LongToken(i), ByteBufferUtil.EMPTY_BYTE_BUFFER), group), val); + double avgSize = ObjectSizes.measureDeep(rows) / (double) count; + rowOverhead = (int) ((avgSize - Math.floor(avgSize)) < 0.05 ? Math.floor(avgSize) : Math.ceil(avgSize)); + rowOverhead -= ObjectSizes.measureDeep(new LongToken(0)); + rowOverhead += AtomicBTreeColumns.EMPTY_SIZE; + allocator.setDiscarding(); + allocator.setDiscarded(); + return rowOverhead; + } } } diff --git a/src/java/org/apache/cassandra/db/MutationVerbHandler.java b/src/java/org/apache/cassandra/db/MutationVerbHandler.java index 92bfdb5cf7..3baa93e927 100644 --- a/src/java/org/apache/cassandra/db/MutationVerbHandler.java +++ b/src/java/org/apache/cassandra/db/MutationVerbHandler.java @@ -59,18 +59,20 @@ public class MutationVerbHandler implements IVerbHandler */ private void forwardToLocalNodes(Mutation mutation, MessagingService.Verb verb, byte[] forwardBytes, InetAddress from) throws IOException { - DataInputStream in = new DataInputStream(new FastByteArrayInputStream(forwardBytes)); - int size = in.readInt(); - - // tell the recipients who to send their ack to - MessageOut message = new MessageOut<>(verb, mutation, Mutation.serializer).withParameter(Mutation.FORWARD_FROM, from.getAddress()); - // Send a message to each of the addresses on our Forward List - for (int i = 0; i < size; i++) + try (DataInputStream in = new DataInputStream(new FastByteArrayInputStream(forwardBytes))) { - InetAddress address = CompactEndpointSerializationHelper.deserialize(in); - int id = in.readInt(); - Tracing.trace("Enqueuing forwarded write to {}", address); - MessagingService.instance().sendOneWay(message, id, address); + int size = in.readInt(); + + // tell the recipients who to send their ack to + MessageOut message = new MessageOut<>(verb, mutation, Mutation.serializer).withParameter(Mutation.FORWARD_FROM, from.getAddress()); + // Send a message to each of the addresses on our Forward List + for (int i = 0; i < size; i++) + { + InetAddress address = CompactEndpointSerializationHelper.deserialize(in); + int id = in.readInt(); + Tracing.trace("Enqueuing forwarded write to {}", address); + MessagingService.instance().sendOneWay(message, id, address); + } } } } diff --git a/src/java/org/apache/cassandra/db/RangeSliceReply.java b/src/java/org/apache/cassandra/db/RangeSliceReply.java index 5964ea8be6..ed1f523a64 100644 --- a/src/java/org/apache/cassandra/db/RangeSliceReply.java +++ b/src/java/org/apache/cassandra/db/RangeSliceReply.java @@ -57,7 +57,10 @@ public class RangeSliceReply public static RangeSliceReply read(byte[] body, int version) throws IOException { - return serializer.deserialize(new DataInputStream(new FastByteArrayInputStream(body)), version); + try (DataInputStream dis = new DataInputStream(new FastByteArrayInputStream(body))) + { + return serializer.deserialize(dis, version); + } } private static class RangeSliceReplySerializer implements IVersionedSerializer diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java index b0c114a3be..c68109ce11 100644 --- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java +++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java @@ -82,6 +82,7 @@ public class SizeEstimatesRecorder extends MigrationListener implements Runnable } } + @SuppressWarnings("resource") private void recordSizeEstimates(ColumnFamilyStore table, Collection> localRanges) { // for each local primary range, estimate (crudely) mean partition size and partitions count. @@ -90,22 +91,24 @@ public class SizeEstimatesRecorder extends MigrationListener implements Runnable { // filter sstables that have partitions in this range. Refs refs = null; - while (refs == null) - { - ColumnFamilyStore.ViewFragment view = table.select(table.viewFilter(Range.makeRowRange(range))); - refs = Refs.tryRef(view.sstables); - } - long partitionsCount, meanPartitionSize; + try { + while (refs == null) + { + ColumnFamilyStore.ViewFragment view = table.select(table.viewFilter(Range.makeRowRange(range))); + refs = Refs.tryRef(view.sstables); + } + // calculate the estimates. partitionsCount = estimatePartitionsCount(refs, range); meanPartitionSize = estimateMeanPartitionSize(refs); } finally { - refs.release(); + if (refs != null) + refs.release(); } estimates.put(range, Pair.create(partitionsCount, meanPartitionSize)); diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 67a3162e0b..9956728aea 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -437,17 +437,16 @@ public final class SystemKeyspace private static Map truncationAsMapEntry(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position) { - DataOutputBuffer out = new DataOutputBuffer(); - try + try (DataOutputBuffer out = new DataOutputBuffer()) { ReplayPosition.serializer.serialize(position, out); out.writeLong(truncatedAt); + return Collections.singletonMap(cfs.metadata.cfId, ByteBuffer.wrap(out.getData(), 0, out.getLength())); } catch (IOException e) { throw new RuntimeException(e); } - return Collections.singletonMap(cfs.metadata.cfId, ByteBuffer.wrap(out.getData(), 0, out.getLength())); } public static ReplayPosition getTruncatedPosition(UUID cfId) @@ -1116,9 +1115,8 @@ public final class SystemKeyspace private static ByteBuffer rangeToBytes(Range range) { - try + try (DataOutputBuffer out = new DataOutputBuffer()) { - DataOutputBuffer out = new DataOutputBuffer(); Range.tokenSerializer.serialize(range, out, MessagingService.VERSION_22); return out.buffer(); } diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index a8dda28d14..a81145d961 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -257,12 +257,10 @@ public class CommitLog implements CommitLogMBean } Allocation alloc = allocator.allocate(mutation, (int) totalSize); - try + ICRC32 checksum = CRC32Factory.instance.create(); + final ByteBuffer buffer = alloc.getBuffer(); + try (BufferedDataOutputStreamPlus dos = new DataOutputBufferFixed(buffer)) { - ICRC32 checksum = CRC32Factory.instance.create(); - final ByteBuffer buffer = alloc.getBuffer(); - BufferedDataOutputStreamPlus dos = new DataOutputBufferFixed(buffer); - // checksummed length dos.writeInt((int) size); checksum.update(buffer, buffer.position() - 4, 4); diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java index 27abae3838..02072de6ab 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java @@ -63,11 +63,8 @@ public class CommitLogArchiver public CommitLogArchiver() { Properties commitlog_commands = new Properties(); - InputStream stream = null; - try + try (InputStream stream = getClass().getClassLoader().getResourceAsStream("commitlog_archiving.properties")) { - stream = getClass().getClassLoader().getResourceAsStream("commitlog_archiving.properties"); - if (stream == null) { logger.debug("No commitlog_archiving properties found; archive + pitr will be disabled"); @@ -113,10 +110,7 @@ public class CommitLogArchiver { throw new RuntimeException("Unable to load commitlog_archiving.properties", e); } - finally - { - FileUtils.closeQuietly(stream); - } + } public void maybeArchive(final CommitLogSegment segment) diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java index 6f9039d822..a59e70e8a4 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java @@ -268,6 +268,7 @@ public class CommitLogReplayer } } + @SuppressWarnings("resource") public void recover(File file) throws IOException { CommitLogDescriptor desc = CommitLogDescriptor.fromFileName(file.getName()); @@ -340,6 +341,7 @@ public class CommitLogReplayer FileDataInput sectionReader = reader; if (compressor != null) + { try { int start = (int) reader.getFilePointer(); @@ -363,6 +365,7 @@ public class CommitLogReplayer logger.error("Unexpected exception decompressing section {}", e); continue; } + } if (!replaySyncSection(sectionReader, replayEnd, desc)) break; @@ -469,9 +472,9 @@ public class CommitLogReplayer void replayMutation(byte[] inputBuffer, int size, final long entryLocation, final CommitLogDescriptor desc) throws IOException { - FastByteArrayInputStream bufIn = new FastByteArrayInputStream(inputBuffer, 0, size); + final Mutation mutation; - try + try (FastByteArrayInputStream bufIn = new FastByteArrayInputStream(inputBuffer, 0, size)) { mutation = Mutation.serializer.deserialize(new DataInputStream(bufIn), desc.getMessagingVersion(), @@ -499,15 +502,12 @@ public class CommitLogReplayer { JVMStabilityInspector.inspectThrowable(t); File f = File.createTempFile("mutation", "dat"); - DataOutputStream out = new DataOutputStream(new FileOutputStream(f)); - try + + try (DataOutputStream out = new DataOutputStream(new FileOutputStream(f))) { out.write(inputBuffer, 0, size); } - finally - { - out.close(); - } + String st = String.format("Unexpected error deserializing mutation; saved to %s and ignored. This may be caused by replaying a mutation against a table with the same name but incompatible schema. Exception follows: ", f.getAbsolutePath()); logger.error(st, t); diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java index d04690dafb..ee160c3b1d 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java @@ -160,6 +160,7 @@ public abstract class CommitLogSegment * Allocate space in this buffer for the provided mutation, and return the allocated Allocation object. * Returns null if there is not enough space in this segment, and a new segment is needed. */ + @SuppressWarnings("resource") //we pass the op order around Allocation allocate(Mutation mutation, int size) { final OpOrder.Group opGroup = appendOrder.start(); diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java index 38107c0c1f..97e7041ebc 100644 --- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java @@ -268,6 +268,7 @@ public abstract class AbstractCompactionStrategy * allow for a more memory efficient solution if we know the sstable don't overlap (see * LeveledCompactionStrategy for instance). */ + @SuppressWarnings("resource") public ScannerList getScanners(Collection sstables, Range range) { RateLimiter limiter = CompactionManager.instance.getRateLimiter(); diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 20793258b7..ffed554d5b 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -42,6 +42,7 @@ import javax.management.openmbean.OpenDataException; import javax.management.openmbean.TabularData; import com.google.common.base.Predicate; +import com.google.common.base.Throwables; import com.google.common.collect.*; import com.google.common.util.concurrent.*; import org.slf4j.Logger; @@ -70,9 +71,11 @@ import org.apache.cassandra.metrics.CompactionMetrics; import org.apache.cassandra.repair.Validator; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.CloseableIterator; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.MerkleTree; +import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.WrappedRunnable; import org.apache.cassandra.utils.UUIDGen; import org.apache.cassandra.utils.concurrent.OpOrder; @@ -243,6 +246,7 @@ public class CompactionManager implements CompactionManagerMBean } } + @SuppressWarnings("resource") private AllSSTableOpStatus parallelAllSSTableOperation(final ColumnFamilyStore cfs, final OneSSTableOperation operation, OperationType operationType) throws ExecutionException, InterruptedException { try (LifecycleTransaction compacting = cfs.markAllCompacting(operationType);) @@ -259,7 +263,8 @@ public class CompactionManager implements CompactionManagerMBean } Iterable sstables = operation.filterSSTables(compacting.originals()); - List> futures = new ArrayList<>(); + List>> futures = new ArrayList<>(); + for (final SSTableReader sstable : sstables) { @@ -270,7 +275,7 @@ public class CompactionManager implements CompactionManagerMBean } final LifecycleTransaction txn = compacting.split(singleton(sstable)); - futures.add(executor.submit(new Callable() + futures.add(Pair.create(txn,executor.submit(new Callable() { @Override public Object call() throws Exception @@ -278,13 +283,37 @@ public class CompactionManager implements CompactionManagerMBean operation.execute(txn); return this; } - })); + }))); } assert compacting.originals().isEmpty(); - for (Future f : futures) - f.get(); + + //Collect all exceptions + Exception exception = null; + + for (Pair> f : futures) + { + try + { + f.right.get(); + } + catch (InterruptedException | ExecutionException e) + { + if (exception == null) + exception = new Exception(); + + exception.addSuppressed(e); + } + finally + { + f.left.close(); + } + } + + if (exception != null) + Throwables.propagate(exception); + return AllSSTableOpStatus.SUCCESSFUL; } } @@ -407,6 +436,7 @@ public class CompactionManager implements CompactionManagerMBean { Runnable runnable = new WrappedRunnable() { @Override + @SuppressWarnings("resource") public void runMayThrow() throws Exception { LifecycleTransaction modifier = null; @@ -427,6 +457,7 @@ public class CompactionManager implements CompactionManagerMBean if (executor.isShutdown()) { logger.info("Compaction executor has shut down, not submitting anticompaction"); + sstables.release(); return Futures.immediateCancelledFuture(); } @@ -659,35 +690,35 @@ public class CompactionManager implements CompactionManagerMBean private void scrubOne(ColumnFamilyStore cfs, LifecycleTransaction modifier, boolean skipCorrupted, boolean checkData) throws IOException { - Scrubber scrubber = new Scrubber(cfs, modifier, skipCorrupted, false, checkData); + CompactionInfo.Holder scrubInfo = null; - CompactionInfo.Holder scrubInfo = scrubber.getScrubInfo(); - metrics.beginCompaction(scrubInfo); - try + try (Scrubber scrubber = new Scrubber(cfs, modifier, skipCorrupted, false, checkData)) { + scrubInfo = scrubber.getScrubInfo(); + metrics.beginCompaction(scrubInfo); scrubber.scrub(); } finally { - scrubber.close(); - metrics.finishCompaction(scrubInfo); + if (scrubInfo != null) + metrics.finishCompaction(scrubInfo); } } private void verifyOne(ColumnFamilyStore cfs, SSTableReader sstable, boolean extendedVerify) throws IOException { - Verifier verifier = new Verifier(cfs, sstable, false); + CompactionInfo.Holder verifyInfo = null; - CompactionInfo.Holder verifyInfo = verifier.getVerifyInfo(); - metrics.beginCompaction(verifyInfo); - try + try (Verifier verifier = new Verifier(cfs, sstable, false)) { + verifyInfo = verifier.getVerifyInfo(); + metrics.beginCompaction(verifyInfo); verifier.verify(extendedVerify); } finally { - verifier.close(); - metrics.finishCompaction(verifyInfo); + if (verifyInfo != null) + metrics.finishCompaction(verifyInfo); } } @@ -798,10 +829,11 @@ public class CompactionManager implements CompactionManagerMBean if (ci.isStopRequested()) throw new CompactionInterruptedException(ci.getCompactionInfo()); - SSTableIdentityIterator row = (SSTableIdentityIterator) scanner.next(); - row = cleanupStrategy.cleanup(row); + @SuppressWarnings("resource") + SSTableIdentityIterator row = cleanupStrategy.cleanup((SSTableIdentityIterator) scanner.next()); if (row == null) continue; + @SuppressWarnings("resource") AbstractCompactedRow compactedRow = new LazilyCompactedRow(controller, Collections.singletonList(row)); if (writer.append(compactedRow) != null) totalkeysWritten++; @@ -991,35 +1023,40 @@ public class CompactionManager implements CompactionManagerMBean if (!cfs.isValid()) return; - Refs sstables = null; - try + String snapshotName = validator.desc.sessionId.toString(); + boolean isSnapshotValidation = cfs.snapshotExists(snapshotName); + + int gcBefore; + + if (isSnapshotValidation) { + // If there is a snapshot created for the session then read from there. + // note that we populate the parent repair session when creating the snapshot, meaning the sstables in the snapshot are the ones we + // are supposed to validate. - String snapshotName = validator.desc.sessionId.toString(); - int gcBefore; - boolean isSnapshotValidation = cfs.snapshotExists(snapshotName); - if (isSnapshotValidation) + try (Refs sstables = cfs.getSnapshotSSTableReader(snapshotName)) { - // If there is a snapshot created for the session then read from there. - // note that we populate the parent repair session when creating the snapshot, meaning the sstables in the snapshot are the ones we - // are supposed to validate. - sstables = cfs.getSnapshotSSTableReader(snapshotName); - - // Computing gcbefore based on the current time wouldn't be very good because we know each replica will execute // this at a different time (that's the whole purpose of repair with snaphsot). So instead we take the creation // time of the snapshot, which should give us roughtly the same time on each replica (roughtly being in that case // 'as good as in the non-snapshot' case) gcBefore = cfs.gcBefore(cfs.getSnapshotCreationTime(snapshotName)); - } - else - { - // flush first so everyone is validating data that is as similar as possible - StorageService.instance.forceKeyspaceFlush(cfs.keyspace.getName(), cfs.name); - ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(validator.desc.parentSessionId); - ColumnFamilyStore.RefViewFragment sstableCandidates = cfs.selectAndReference(prs.isIncremental ? ColumnFamilyStore.UNREPAIRED_SSTABLES : ColumnFamilyStore.CANONICAL_SSTABLES); - Set sstablesToValidate = new HashSet<>(); + buildMerkleTree(cfs, sstables, validator, gcBefore); + + // review comment: should this be in a try/finally? it was previously + cfs.clearSnapshot(snapshotName); + } + } + else + { + // flush first so everyone is validating data that is as similar as possible + StorageService.instance.forceKeyspaceFlush(cfs.keyspace.getName(), cfs.name); + ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(validator.desc.parentSessionId); + try (ColumnFamilyStore.RefViewFragment sstableCandidates = cfs.selectAndReference(prs.isIncremental ? ColumnFamilyStore.UNREPAIRED_SSTABLES : ColumnFamilyStore.CANONICAL_SSTABLES)) + { + Refs refs = sstableCandidates.refs; + Set sstablesToValidate = new HashSet<>(); for (SSTableReader sstable : sstableCandidates.sstables) { if (new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(Collections.singletonList(validator.desc.range))) @@ -1036,82 +1073,74 @@ public class CompactionManager implements CompactionManagerMBean throw new RuntimeException("Cannot start multiple repair sessions over the same sstables"); } - sstables = Refs.tryRef(sstablesToValidate); - if (sstables == null) - { - logger.error("Could not reference sstables"); - throw new RuntimeException("Could not reference sstables"); - } - sstableCandidates.release(); + refs.relaseAllExcept(sstablesToValidate); prs.addSSTables(cfs.metadata.cfId, sstablesToValidate); if (validator.gcBefore > 0) gcBefore = validator.gcBefore; else gcBefore = getDefaultGcBefore(cfs); - } - // Create Merkle tree suitable to hold estimated partitions for given range. - // We blindly assume that partition is evenly distributed on all sstables for now. - long numPartitions = 0; - for (SSTableReader sstable : sstables) - { - numPartitions += sstable.estimatedKeysForRanges(singleton(validator.desc.range)); - } - // determine tree depth from number of partitions, but cap at 20 to prevent large tree. - int depth = numPartitions > 0 ? (int) Math.min(Math.floor(Math.log(numPartitions)), 20) : 0; - MerkleTree tree = new MerkleTree(cfs.partitioner, validator.desc.range, MerkleTree.RECOMMENDED_DEPTH, (int) Math.pow(2, depth)); - long start = System.nanoTime(); - try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables, validator.desc.range)) - { - CompactionIterable ci = new ValidationCompactionIterable(cfs, scanners.scanners, gcBefore); - Iterator iter = ci.iterator(); - metrics.beginCompaction(ci); - try - { - // validate the CF as we iterate over it - validator.prepare(cfs, tree); - while (iter.hasNext()) - { - if (ci.isStopRequested()) - throw new CompactionInterruptedException(ci.getCompactionInfo()); - AbstractCompactedRow row = iter.next(); - validator.add(row); - } - validator.complete(); - } - finally - { - if (isSnapshotValidation) - { - cfs.clearSnapshot(snapshotName); - } - - metrics.finishCompaction(ci); - } + buildMerkleTree(cfs, refs, validator, gcBefore); } - - if (logger.isDebugEnabled()) - { - // MT serialize may take time - long duration = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); - logger.debug("Validation finished in {} msec, depth {} for {} keys, serialized size {} bytes for {}", - duration, - depth, - numPartitions, - MerkleTree.serializer.serializedSize(tree, 0), - validator.desc); - } - } - finally - { - if (sstables != null) - sstables.release(); } } + private void buildMerkleTree(ColumnFamilyStore cfs, Refs sstables, Validator validator, int gcBefore) + { + // Create Merkle tree suitable to hold estimated partitions for given range. + // We blindly assume that partition is evenly distributed on all sstables for now. + long numPartitions = 0; + for (SSTableReader sstable : sstables) + { + numPartitions += sstable.estimatedKeysForRanges(Collections.singleton(validator.desc.range)); + } + // determine tree depth from number of partitions, but cap at 20 to prevent large tree. + int depth = numPartitions > 0 ? (int) Math.min(Math.floor(Math.log(numPartitions)), 20) : 0; + MerkleTree tree = new MerkleTree(cfs.partitioner, validator.desc.range, MerkleTree.RECOMMENDED_DEPTH, (int) Math.pow(2, depth)); + long start = System.nanoTime(); + try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables, validator.desc.range)) + { + CompactionIterable ci = new ValidationCompactionIterable(cfs, scanners.scanners, gcBefore); + metrics.beginCompaction(ci); + try (CloseableIterator iter = ci.iterator();) + { + // validate the CF as we iterate over it + validator.prepare(cfs, tree); + while (iter.hasNext()) + { + if (ci.isStopRequested()) + throw new CompactionInterruptedException(ci.getCompactionInfo()); + @SuppressWarnings("resource") + AbstractCompactedRow row = iter.next(); + validator.add(row); + } + validator.complete(); + } + catch (Exception e) + { + Throwables.propagate(e); + } + finally + { + metrics.finishCompaction(ci); + } + } + + if (logger.isDebugEnabled()) + { + // MT serialize may take time + long duration = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + logger.debug("Validation finished in {} msec, depth {} for {} keys, serialized size {} bytes for {}", + duration, + depth, + numPartitions, + MerkleTree.serializer.serializedSize(tree, 0), + validator.desc); + } + } /** * Splits up an sstable into two new sstables. The first of the new tables will store repaired ranges, the second @@ -1187,12 +1216,14 @@ public class CompactionManager implements CompactionManagerMBean unRepairedSSTableWriter.switchWriter(CompactionManager.createWriterForAntiCompaction(cfs, destination, expectedBloomFilterSize, ActiveRepairService.UNREPAIRED_SSTABLE, sstableAsSet)); CompactionIterable ci = new CompactionIterable(OperationType.ANTICOMPACTION, scanners.scanners, controller, DatabaseDescriptor.getSSTableFormat(), UUIDGen.getTimeUUID()); - Iterator iter = ci.iterator(); metrics.beginCompaction(ci); try { + @SuppressWarnings("resource") + CloseableIterator iter = ci.iterator(); while (iter.hasNext()) { + @SuppressWarnings("resource") AbstractCompactedRow row = iter.next(); // if current range from sstable is repaired, save it into the new repaired sstable if (Range.isInRanges(row.key.getToken(), ranges)) @@ -1315,6 +1346,7 @@ public class CompactionManager implements CompactionManagerMBean private static class ValidationCompactionIterable extends CompactionIterable { + @SuppressWarnings("resource") public ValidationCompactionIterable(ColumnFamilyStore cfs, List scanners, int gcBefore) { super(OperationType.VALIDATION, scanners, new ValidationCompactionController(cfs, gcBefore), DatabaseDescriptor.getSSTableFormat(), UUIDGen.getTimeUUID()); diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java index e593ec0b45..70890169ab 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java @@ -17,9 +17,9 @@ */ package org.apache.cassandra.db.compaction; +import java.util.Arrays; import java.util.Collection; import java.util.HashMap; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; @@ -44,6 +44,7 @@ import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.compaction.CompactionManager.CompactionExecutorStatsCollector; import org.apache.cassandra.db.lifecycle.LifecycleTransaction; import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.utils.CloseableIterator; import org.apache.cassandra.utils.UUIDGen; public class CompactionTask extends AbstractCompactionTask @@ -159,45 +160,49 @@ public class CompactionTask extends AbstractCompactionTask try (AbstractCompactionStrategy.ScannerList scanners = strategy.getScanners(actuallyCompact)) { ci = new CompactionIterable(compactionType, scanners.scanners, controller, sstableFormat, taskId); - Iterator iter = ci.iterator(); - if (collector != null) - collector.beginCompaction(ci); - long lastCheckObsoletion = start; - - if (!controller.cfs.getCompactionStrategy().isActive) - throw new CompactionInterruptedException(ci.getCompactionInfo()); - - try (CompactionAwareWriter writer = getCompactionAwareWriter(cfs, transaction, actuallyCompact)) + try (CloseableIterator iter = ci.iterator()) { - estimatedKeys = writer.estimatedKeys(); - while (iter.hasNext()) - { - if (ci.isStopRequested()) - throw new CompactionInterruptedException(ci.getCompactionInfo()); - - AbstractCompactedRow row = iter.next(); - if (writer.append(row)) - totalKeysWritten++; - - if (System.nanoTime() - lastCheckObsoletion > TimeUnit.MINUTES.toNanos(1L)) - { - controller.maybeRefreshOverlaps(); - lastCheckObsoletion = System.nanoTime(); - } - } - - // don't replace old sstables yet, as we need to mark the compaction finished in the system table - newSStables = writer.finish(); - } - finally - { - // point of no return -- the new sstables are live on disk; next we'll start deleting the old ones - // (in replaceCompactedSSTables) - if (taskId != null) - SystemKeyspace.finishCompaction(taskId); - if (collector != null) - collector.finishCompaction(ci); + collector.beginCompaction(ci); + long lastCheckObsoletion = start; + + if (!controller.cfs.getCompactionStrategy().isActive) + throw new CompactionInterruptedException(ci.getCompactionInfo()); + + try (CompactionAwareWriter writer = getCompactionAwareWriter(cfs, transaction, actuallyCompact)) + { + estimatedKeys = writer.estimatedKeys(); + while (iter.hasNext()) + { + if (ci.isStopRequested()) + throw new CompactionInterruptedException(ci.getCompactionInfo()); + + try (AbstractCompactedRow row = iter.next()) + { + if (writer.append(row)) + totalKeysWritten++; + + if (System.nanoTime() - lastCheckObsoletion > TimeUnit.MINUTES.toNanos(1L)) + { + controller.maybeRefreshOverlaps(); + lastCheckObsoletion = System.nanoTime(); + } + } + } + + // don't replace old sstables yet, as we need to mark the compaction finished in the system table + newSStables = writer.finish(); + } + finally + { + // point of no return -- the new sstables are live on disk; next we'll start deleting the old ones + // (in replaceCompactedSSTables) + if (taskId != null) + SystemKeyspace.finishCompaction(taskId); + + if (collector != null) + collector.finishCompaction(ci); + } } } diff --git a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java index 18d5f7b3da..43f998aae0 100644 --- a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java @@ -56,6 +56,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy } @Override + @SuppressWarnings("resource") public synchronized AbstractCompactionTask getNextBackgroundTask(int gcBefore) { if (!isEnabled()) @@ -366,6 +367,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy } @Override + @SuppressWarnings("resource") public synchronized Collection getMaximalTask(int gcBefore, boolean splitOutput) { LifecycleTransaction modifier = cfs.markAllCompacting(OperationType.COMPACTION); @@ -376,6 +378,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy } @Override + @SuppressWarnings("resource") public synchronized AbstractCompactionTask getUserDefinedTask(Collection sstables, int gcBefore) { assert !sstables.isEmpty(); // checked for by CM.submitUserDefined diff --git a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java index c3d764eb0f..9eb624e63c 100644 --- a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java +++ b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java @@ -39,6 +39,7 @@ import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.io.util.SequentialWriter; import org.apache.cassandra.utils.MergeIterator; import org.apache.cassandra.utils.StreamingHistogram; +import org.apache.cassandra.utils.Throwables; /** * LazilyCompactedRow only computes the row bloom filter and column index in memory @@ -157,13 +158,11 @@ public class LazilyCompactedRow extends AbstractCompactedRow // no special-case for rows.size == 1, we're actually skipping some bytes here so just // blindly updating everything wouldn't be correct - DataOutputBuffer out = new DataOutputBuffer(); - - // initialize indexBuilder for the benefit of its tombstoneTracker, used by our reducing iterator - indexBuilder = new ColumnIndex.Builder(emptyColumnFamily, key.getKey(), out); - - try + try (DataOutputBuffer out = new DataOutputBuffer()) { + // initialize indexBuilder for the benefit of its tombstoneTracker, used by our reducing iterator + indexBuilder = new ColumnIndex.Builder(emptyColumnFamily, key.getKey(), out); + DeletionTime.serializer.serialize(emptyColumnFamily.deletionInfo().getTopLevelDeletion(), out); // do not update digest in case of missing or purged row level tombstones, see CASSANDRA-8979 @@ -192,6 +191,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow public void close() { + Throwable accumulate = null; for (OnDiskAtomIterator row : rows) { try @@ -200,10 +200,11 @@ public class LazilyCompactedRow extends AbstractCompactedRow } catch (IOException e) { - throw new RuntimeException(e); + accumulate = Throwables.merge(accumulate, e); } } closed = true; + Throwables.maybeFail(accumulate); } protected class Reducer extends MergeIterator.Reducer diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java index c434d31014..9eb58ff3df 100644 --- a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java @@ -88,6 +88,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy * the only difference between background and maximal in LCS is that maximal is still allowed * (by explicit user request) even when compaction is disabled. */ + @SuppressWarnings("resource") public synchronized AbstractCompactionTask getNextBackgroundTask(int gcBefore) { if (!isEnabled()) @@ -126,6 +127,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy } } + @SuppressWarnings("resource") public synchronized Collection getMaximalTask(int gcBefore, boolean splitOutput) { Iterable sstables = manifest.getAllSSTables(); diff --git a/src/java/org/apache/cassandra/db/compaction/Scrubber.java b/src/java/org/apache/cassandra/db/compaction/Scrubber.java index 93b76bdcde..10952e7b2d 100644 --- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java +++ b/src/java/org/apache/cassandra/db/compaction/Scrubber.java @@ -86,6 +86,7 @@ public class Scrubber implements Closeable this(cfs, transaction, skipCorrupted, new OutputHandler.LogOutput(), isOffline, checkData); } + @SuppressWarnings("resource") public Scrubber(ColumnFamilyStore cfs, LifecycleTransaction transaction, boolean skipCorrupted, OutputHandler outputHandler, boolean isOffline, boolean checkData) throws IOException { this.cfs = cfs; @@ -204,6 +205,7 @@ public class Scrubber implements Closeable continue; } + @SuppressWarnings("resource") AbstractCompactedRow compactedRow = new LazilyCompactedRow(controller, Collections.singletonList(atoms)); if (writer.tryAppend(compactedRow) == null) emptyRows++; @@ -234,6 +236,7 @@ public class Scrubber implements Closeable continue; } + @SuppressWarnings("resource") AbstractCompactedRow compactedRow = new LazilyCompactedRow(controller, Collections.singletonList(atoms)); if (writer.tryAppend(compactedRow) == null) emptyRows++; diff --git a/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java index 94c3daf09d..74a975788a 100644 --- a/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java @@ -175,6 +175,7 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy return sstr.getReadMeter() == null ? 0.0 : sstr.getReadMeter().twoHourRate() / sstr.estimatedKeys(); } + @SuppressWarnings("resource") public synchronized AbstractCompactionTask getNextBackgroundTask(int gcBefore) { if (!isEnabled()) @@ -193,6 +194,7 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy } } + @SuppressWarnings("resource") public Collection getMaximalTask(final int gcBefore, boolean splitOutput) { Iterable filteredSSTables = filterSuspectSSTables(sstables); @@ -206,6 +208,7 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy return Arrays.asList(new CompactionTask(cfs, txn, gcBefore, false)); } + @SuppressWarnings("resource") public AbstractCompactionTask getUserDefinedTask(Collection sstables, final int gcBefore) { assert !sstables.isEmpty(); // checked for by CM.submitUserDefined diff --git a/src/java/org/apache/cassandra/db/compaction/Upgrader.java b/src/java/org/apache/cassandra/db/compaction/Upgrader.java index 6556a71da5..ca975b8936 100644 --- a/src/java/org/apache/cassandra/db/compaction/Upgrader.java +++ b/src/java/org/apache/cassandra/db/compaction/Upgrader.java @@ -20,6 +20,8 @@ package org.apache.cassandra.db.compaction; import java.io.File; import java.util.*; +import com.google.common.base.Throwables; + import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.DecoratedKey; @@ -28,6 +30,7 @@ import org.apache.cassandra.io.sstable.*; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.sstable.format.SSTableWriter; import org.apache.cassandra.io.sstable.metadata.MetadataCollector; +import org.apache.cassandra.utils.CloseableIterator; import org.apache.cassandra.utils.OutputHandler; import org.apache.cassandra.utils.UUIDGen; @@ -83,12 +86,13 @@ public class Upgrader outputHandler.output("Upgrading " + sstable); try (SSTableRewriter writer = new SSTableRewriter(cfs, transaction, CompactionTask.getMaxDataAge(transaction.originals()), true); - AbstractCompactionStrategy.ScannerList scanners = strategy.getScanners(transaction.originals())) + AbstractCompactionStrategy.ScannerList scanners = strategy.getScanners(transaction.originals()); + CloseableIterator iter = new CompactionIterable(compactionType, scanners.scanners, controller, DatabaseDescriptor.getSSTableFormat(), UUIDGen.getTimeUUID()).iterator()) { - Iterator iter = new CompactionIterable(compactionType, scanners.scanners, controller, DatabaseDescriptor.getSSTableFormat(), UUIDGen.getTimeUUID()).iterator(); writer.switchWriter(createCompactionWriter(sstable.getSSTableMetadata().repairedAt)); while (iter.hasNext()) { + @SuppressWarnings("resource") AbstractCompactedRow row = iter.next(); writer.append(row); } @@ -96,6 +100,10 @@ public class Upgrader writer.finish(); outputHandler.output("Upgrade of " + sstable + " complete."); } + catch (Exception e) + { + Throwables.propagate(e); + } finally { controller.close(); diff --git a/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java index c511bcd949..adda0c9b38 100644 --- a/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java @@ -344,6 +344,7 @@ public final class WrappingCompactionStrategy extends AbstractCompactionStrategy } @Override + @SuppressWarnings("resource") public synchronized ScannerList getScanners(Collection sstables, Range range) { List repairedSSTables = new ArrayList<>(); diff --git a/src/java/org/apache/cassandra/db/compaction/writers/DefaultCompactionWriter.java b/src/java/org/apache/cassandra/db/compaction/writers/DefaultCompactionWriter.java index 0b3106148d..7d88458206 100644 --- a/src/java/org/apache/cassandra/db/compaction/writers/DefaultCompactionWriter.java +++ b/src/java/org/apache/cassandra/db/compaction/writers/DefaultCompactionWriter.java @@ -41,12 +41,14 @@ public class DefaultCompactionWriter extends CompactionAwareWriter { protected static final Logger logger = LoggerFactory.getLogger(DefaultCompactionWriter.class); + @SuppressWarnings("resource") public DefaultCompactionWriter(ColumnFamilyStore cfs, LifecycleTransaction txn, Set nonExpiredSSTables, boolean offline, OperationType compactionType) { super(cfs, txn, nonExpiredSSTables, offline); logger.debug("Expected bloom filter size : {}", estimatedTotalKeys); long expectedWriteSize = cfs.getExpectedCompactedFileSize(nonExpiredSSTables, compactionType); File sstableDirectory = cfs.directories.getLocationForDisk(getWriteDirectory(expectedWriteSize)); + @SuppressWarnings("resource") SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(cfs.getTempSSTablePath(sstableDirectory)), estimatedTotalKeys, minRepairedAt, diff --git a/src/java/org/apache/cassandra/db/compaction/writers/MajorLeveledCompactionWriter.java b/src/java/org/apache/cassandra/db/compaction/writers/MajorLeveledCompactionWriter.java index 014b4af046..95d7a0c82e 100644 --- a/src/java/org/apache/cassandra/db/compaction/writers/MajorLeveledCompactionWriter.java +++ b/src/java/org/apache/cassandra/db/compaction/writers/MajorLeveledCompactionWriter.java @@ -47,6 +47,7 @@ public class MajorLeveledCompactionWriter extends CompactionAwareWriter private int sstablesWritten = 0; private final boolean skipAncestors; + @SuppressWarnings("resource") public MajorLeveledCompactionWriter(ColumnFamilyStore cfs, LifecycleTransaction txn, Set nonExpiredSSTables, long maxSSTableSize, boolean offline, OperationType compactionType) { super(cfs, txn, nonExpiredSSTables, offline); @@ -61,6 +62,7 @@ public class MajorLeveledCompactionWriter extends CompactionAwareWriter if (skipAncestors) logger.warn("Many sstables involved in compaction, skipping storing ancestor information to avoid running out of memory"); + @SuppressWarnings("resource") SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(cfs.getTempSSTablePath(sstableDirectory)), keysPerSSTable, minRepairedAt, @@ -71,6 +73,7 @@ public class MajorLeveledCompactionWriter extends CompactionAwareWriter } @Override + @SuppressWarnings("resource") public boolean append(AbstractCompactedRow row) { long posBefore = sstableWriter.currentWriter().getOnDiskFilePointer(); diff --git a/src/java/org/apache/cassandra/db/compaction/writers/MaxSSTableSizeWriter.java b/src/java/org/apache/cassandra/db/compaction/writers/MaxSSTableSizeWriter.java index 8903ff7a72..d30a61208e 100644 --- a/src/java/org/apache/cassandra/db/compaction/writers/MaxSSTableSizeWriter.java +++ b/src/java/org/apache/cassandra/db/compaction/writers/MaxSSTableSizeWriter.java @@ -39,6 +39,7 @@ public class MaxSSTableSizeWriter extends CompactionAwareWriter private final long estimatedSSTables; private final Set allSSTables; + @SuppressWarnings("resource") public MaxSSTableSizeWriter(ColumnFamilyStore cfs, LifecycleTransaction txn, Set nonExpiredSSTables, long maxSSTableSize, int level, boolean offline, OperationType compactionType) { super(cfs, txn, nonExpiredSSTables, offline); @@ -50,6 +51,7 @@ public class MaxSSTableSizeWriter extends CompactionAwareWriter estimatedTotalKeys = SSTableReader.getApproximateKeyCount(nonExpiredSSTables); estimatedSSTables = Math.max(1, estimatedTotalKeys / maxSSTableSize); File sstableDirectory = cfs.directories.getLocationForDisk(getWriteDirectory(expectedWriteSize)); + @SuppressWarnings("resource") SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(cfs.getTempSSTablePath(sstableDirectory)), estimatedTotalKeys / estimatedSSTables, minRepairedAt, @@ -66,6 +68,7 @@ public class MaxSSTableSizeWriter extends CompactionAwareWriter if (sstableWriter.currentWriter().getOnDiskFilePointer() > maxSSTableSize) { File sstableDirectory = cfs.directories.getLocationForDisk(getWriteDirectory(expectedWriteSize)); + @SuppressWarnings("resource") SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(cfs.getTempSSTablePath(sstableDirectory)), estimatedTotalKeys / estimatedSSTables, minRepairedAt, diff --git a/src/java/org/apache/cassandra/db/compaction/writers/SplittingSizeTieredCompactionWriter.java b/src/java/org/apache/cassandra/db/compaction/writers/SplittingSizeTieredCompactionWriter.java index 81ea6b19d9..9ff1325b49 100644 --- a/src/java/org/apache/cassandra/db/compaction/writers/SplittingSizeTieredCompactionWriter.java +++ b/src/java/org/apache/cassandra/db/compaction/writers/SplittingSizeTieredCompactionWriter.java @@ -56,6 +56,7 @@ public class SplittingSizeTieredCompactionWriter extends CompactionAwareWriter this(cfs, txn, nonExpiredSSTables, compactionType, DEFAULT_SMALLEST_SSTABLE_BYTES); } + @SuppressWarnings("resource") public SplittingSizeTieredCompactionWriter(ColumnFamilyStore cfs, LifecycleTransaction txn, Set nonExpiredSSTables, OperationType compactionType, long smallestSSTable) { super(cfs, txn, nonExpiredSSTables, false); @@ -83,6 +84,7 @@ public class SplittingSizeTieredCompactionWriter extends CompactionAwareWriter File sstableDirectory = cfs.directories.getLocationForDisk(getWriteDirectory(Math.round(totalSize * ratios[currentRatioIndex]))); long currentPartitionsToWrite = Math.round(estimatedTotalKeys * ratios[currentRatioIndex]); currentBytesToWrite = Math.round(totalSize * ratios[currentRatioIndex]); + @SuppressWarnings("resource") SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(cfs.getTempSSTablePath(sstableDirectory)), currentPartitionsToWrite, minRepairedAt, @@ -104,6 +106,7 @@ public class SplittingSizeTieredCompactionWriter extends CompactionAwareWriter currentBytesToWrite = Math.round(totalSize * ratios[currentRatioIndex]); long currentPartitionsToWrite = Math.round(ratios[currentRatioIndex] * estimatedTotalKeys); File sstableDirectory = cfs.directories.getLocationForDisk(getWriteDirectory(Math.round(totalSize * ratios[currentRatioIndex]))); + @SuppressWarnings("resource") SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(cfs.getTempSSTablePath(sstableDirectory)), currentPartitionsToWrite, minRepairedAt, diff --git a/src/java/org/apache/cassandra/db/marshal/CompositeType.java b/src/java/org/apache/cassandra/db/marshal/CompositeType.java index 1bc772d62e..f3c041ef22 100644 --- a/src/java/org/apache/cassandra/db/marshal/CompositeType.java +++ b/src/java/org/apache/cassandra/db/marshal/CompositeType.java @@ -402,9 +402,8 @@ public class CompositeType extends AbstractCompositeType public ByteBuffer build() { - try + try (DataOutputBuffer out = new DataOutputBufferFixed(serializedSize)) { - DataOutputBuffer out = new DataOutputBufferFixed(serializedSize); if (isStatic) out.writeShort(STATIC_MARKER); diff --git a/src/java/org/apache/cassandra/gms/FailureDetector.java b/src/java/org/apache/cassandra/gms/FailureDetector.java index b8c20d76f6..45867a37f7 100644 --- a/src/java/org/apache/cassandra/gms/FailureDetector.java +++ b/src/java/org/apache/cassandra/gms/FailureDetector.java @@ -177,20 +177,14 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean { File file = FileUtils.createTempFile("failuredetector-", ".dat"); - OutputStream os = null; - try + try (OutputStream os = new BufferedOutputStream(new FileOutputStream(file, true))) { - os = new BufferedOutputStream(new FileOutputStream(file, true)); os.write(toString().getBytes()); } catch (IOException e) { throw new FSWriteError(e, file); } - finally - { - FileUtils.closeQuietly(os); - } } public void setPhiConvictThreshold(double phi) diff --git a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java index 2c04475191..4dd53ff28f 100644 --- a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java +++ b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java @@ -241,13 +241,14 @@ public abstract class AbstractColumnFamilyInputFormat extends InputFormat< private Map> getRangeMap(Configuration conf, String keyspace) { - Session session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect(); - - Map> map = new HashMap<>(); - Metadata metadata = session.getCluster().getMetadata(); - for (TokenRange tokenRange : metadata.getTokenRanges()) - map.put(tokenRange, metadata.getReplicas('"' + keyspace + '"', tokenRange)); - return map; + try (Session session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect()) + { + Map> map = new HashMap<>(); + Metadata metadata = session.getCluster().getMetadata(); + for (TokenRange tokenRange : metadata.getTokenRanges()) + map.put(tokenRange, metadata.getReplicas('"' + keyspace + '"', tokenRange)); + return map; + } } private Map describeSplits(String keyspace, String table, TokenRange tokenRange, int splitSize) diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java index 88dd2e2d35..f89825f2f9 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java @@ -61,6 +61,7 @@ public class ColumnFamilyInputFormat extends AbstractColumnFamilyInputFormat, List, List(); - metadata = session.getCluster().getMetadata(); - Set ranges = metadata.getTokenRanges(); - for (TokenRange range : ranges) + try (Session session = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf).connect(keyspace)) { - rangeMap.put(range, metadata.getReplicas(keyspace, range)); + rangeMap = new HashMap<>(); + metadata = session.getCluster().getMetadata(); + Set ranges = metadata.getTokenRanges(); + for (TokenRange range : ranges) + { + rangeMap.put(range, metadata.getReplicas(keyspace, range)); + } } } diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java index 0b833b78da..63baf9c4a8 100644 --- a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java +++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java @@ -699,9 +699,8 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo // Only get the schema if we haven't already gotten it if (!properties.containsKey(signature)) { - try + try (Session client = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf), conf).connect()) { - Session client = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf), conf).connect(); client.execute("USE " + keyspace); // compose the CfDef for the columfamily @@ -729,9 +728,11 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo { TableInfo tableInfo = new TableInfo(cfDef); ByteArrayOutputStream baos = new ByteArrayOutputStream(); - ObjectOutputStream oos = new ObjectOutputStream( baos ); - oos.writeObject( tableInfo ); - oos.close(); + try (ObjectOutputStream oos = new ObjectOutputStream( baos )) + { + oos.writeObject(tableInfo); + } + return new String( Base64Coder.encode(baos.toByteArray()) ); } @@ -739,11 +740,11 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo protected static TableInfo cfdefFromString(String st) throws IOException, ClassNotFoundException { byte [] data = Base64Coder.decode( st ); - ObjectInputStream ois = new ObjectInputStream( - new ByteArrayInputStream( data ) ); - Object o = ois.readObject(); - ois.close(); - return (TableInfo)o; + try (ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(data))) + { + Object o = ois.readObject(); + return (TableInfo)o; + } } /** decompose the query to store the parameters in a map */ diff --git a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java index c994a3dcac..23a9f3e48c 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java +++ b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java @@ -91,17 +91,7 @@ public class CompressionMetadata { this.indexFilePath = indexFilePath; - DataInputStream stream; - try - { - stream = new DataInputStream(new FileInputStream(indexFilePath)); - } - catch (FileNotFoundException e) - { - throw new RuntimeException(e); - } - - try + try (DataInputStream stream = new DataInputStream(new FileInputStream(indexFilePath))) { String compressorName = stream.readUTF(); int optionCount = stream.readInt(); @@ -126,14 +116,14 @@ public class CompressionMetadata compressedFileLength = compressedLength; chunkOffsets = readChunkOffsets(stream); } + catch (FileNotFoundException e) + { + throw new RuntimeException(e); + } catch (IOException e) { throw new CorruptSSTableException(e, indexFilePath); } - finally - { - FileUtils.closeQuietly(stream); - } this.chunkOffsetsSize = chunkOffsets.size(); } @@ -176,32 +166,42 @@ public class CompressionMetadata */ private Memory readChunkOffsets(DataInput input) { + final int chunkCount; try { - int chunkCount = input.readInt(); + chunkCount = input.readInt(); if (chunkCount <= 0) throw new IOException("Compressed file with 0 chunks encountered: " + input); + } + catch (IOException e) + { + throw new FSReadError(e, indexFilePath); + } - Memory offsets = Memory.allocate(chunkCount * 8L); + @SuppressWarnings("resource") + Memory offsets = Memory.allocate(chunkCount * 8L); + int i = 0; + try + { - for (int i = 0; i < chunkCount; i++) + for (i = 0; i < chunkCount; i++) { - try - { - offsets.setLong(i * 8L, input.readLong()); - } - catch (EOFException e) - { - String msg = String.format("Corrupted Index File %s: read %d but expected %d chunks.", - indexFilePath, i, chunkCount); - throw new CorruptSSTableException(new IOException(msg, e), indexFilePath); - } + offsets.setLong(i * 8L, input.readLong()); } return offsets; } catch (IOException e) { + if (offsets != null) + offsets.close(); + + if (e instanceof EOFException) + { + String msg = String.format("Corrupted Index File %s: read %d but expected %d chunks.", + indexFilePath, i, chunkCount); + throw new CorruptSSTableException(new IOException(msg, e), indexFilePath); + } throw new FSReadError(e, indexFilePath); } } @@ -345,10 +345,8 @@ public class CompressionMetadata } // flush the data to disk - DataOutputStream out = null; - try + try (DataOutputStream out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(filePath)))) { - out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(filePath))); writeHeader(out, dataLength, count); for (int i = 0 ; i < count ; i++) out.writeLong(offsets.getLong(i * 8L)); @@ -357,12 +355,9 @@ public class CompressionMetadata { throw Throwables.propagate(e); } - finally - { - FileUtils.closeQuietly(out); - } } + @SuppressWarnings("resource") public CompressionMetadata open(long dataLength, long compressedLength) { SafeMemory offsets = this.offsets.sharedCopy(); diff --git a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java index 1389ad2a88..4181ed050a 100644 --- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java @@ -509,6 +509,7 @@ public class CQLSSTableWriter implements Closeable } } + @SuppressWarnings("resource") public CQLSSTableWriter build() { if (directory == null) diff --git a/src/java/org/apache/cassandra/io/sstable/IndexSummary.java b/src/java/org/apache/cassandra/io/sstable/IndexSummary.java index 59c5eef773..7df734928d 100644 --- a/src/java/org/apache/cassandra/io/sstable/IndexSummary.java +++ b/src/java/org/apache/cassandra/io/sstable/IndexSummary.java @@ -289,6 +289,7 @@ public class IndexSummary extends WrappedSharedCloseable out.write(t.entries, 0, t.entriesLength); } + @SuppressWarnings("resource") public IndexSummary deserialize(DataInputStream in, IPartitioner partitioner, boolean haveSamplingLevel, int expectedMinIndexInterval, int maxIndexInterval) throws IOException { int minIndexInterval = in.readInt(); @@ -321,8 +322,17 @@ public class IndexSummary extends WrappedSharedCloseable Memory offsets = Memory.allocate(offsetCount * 4); Memory entries = Memory.allocate(offheapSize - offsets.size()); - FBUtilities.copy(in, new MemoryOutputStream(offsets), offsets.size()); - FBUtilities.copy(in, new MemoryOutputStream(entries), entries.size()); + try + { + FBUtilities.copy(in, new MemoryOutputStream(offsets), offsets.size()); + FBUtilities.copy(in, new MemoryOutputStream(entries), entries.size()); + } + catch (IOException ioe) + { + offsets.free(); + entries.free(); + throw ioe; + } // our on-disk representation treats the offsets and the summary data as one contiguous structure, // in which the offsets are based from the start of the structure. i.e., if the offsets occupy // X bytes, the value of the first offset will be X. In memory we split the two regions up, so that diff --git a/src/java/org/apache/cassandra/io/sstable/IndexSummaryBuilder.java b/src/java/org/apache/cassandra/io/sstable/IndexSummaryBuilder.java index 12e41c8175..6110afe937 100644 --- a/src/java/org/apache/cassandra/io/sstable/IndexSummaryBuilder.java +++ b/src/java/org/apache/cassandra/io/sstable/IndexSummaryBuilder.java @@ -281,6 +281,7 @@ public class IndexSummaryBuilder implements AutoCloseable * @param partitioner the partitioner used for the index summary * @return a new IndexSummary */ + @SuppressWarnings("resource") public static IndexSummary downsample(IndexSummary existing, int newSamplingLevel, int minIndexInterval, IPartitioner partitioner) { // To downsample the old index summary, we'll go through (potentially) several rounds of downsampling. diff --git a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java index 9bfbc9903b..6f66fd3491 100644 --- a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java +++ b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java @@ -202,6 +202,7 @@ public class IndexSummaryManager implements IndexSummaryManagerMBean * Returns a Pair of all compacting and non-compacting sstables. Non-compacting sstables will be marked as * compacting. */ + @SuppressWarnings("resource") private Pair, Map> getCompactingAndNonCompactingSSTables() { List allCompacting = new ArrayList<>(); diff --git a/src/java/org/apache/cassandra/io/sstable/SSTable.java b/src/java/org/apache/cassandra/io/sstable/SSTable.java index bc3486a17c..20771522de 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTable.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTable.java @@ -270,10 +270,8 @@ public abstract class SSTable protected static void appendTOC(Descriptor descriptor, Collection components) { File tocFile = new File(descriptor.filenameFor(Component.TOC)); - PrintWriter w = null; - try + try (PrintWriter w = new PrintWriter(new FileWriter(tocFile, true))) { - w = new PrintWriter(new FileWriter(tocFile, true)); for (Component component : components) w.println(component.name); } @@ -281,10 +279,6 @@ public abstract class SSTable { throw new FSWriteError(e, tocFile); } - finally - { - FileUtils.closeQuietly(w); - } } /** diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java index 15008d2d8f..b99003b3cb 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java @@ -71,6 +71,7 @@ public class SSTableLoader implements StreamEventHandler this.connectionsPerHost = connectionsPerHost; } + @SuppressWarnings("resource") protected Collection openSSTables(final Map>> ranges) { outputHandler.output("Opening sstables and calculating sections to stream"); @@ -126,9 +127,7 @@ public class SSTableLoader implements StreamEventHandler List> sstableSections = sstable.getPositionsForRanges(tokenRanges); long estimatedKeys = sstable.estimatedKeysForRanges(tokenRanges); - Ref ref = sstable.tryRef(); - if (ref == null) - throw new IllegalStateException("Could not acquire ref for "+sstable); + Ref ref = sstable.ref(); StreamSession.SSTableStreamingSections details = new StreamSession.SSTableStreamingSections(ref, sstableSections, estimatedKeys, ActiveRepairService.UNREPAIRED_SSTABLE); streamingDetails.put(endpoint, details); } diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java index 8e701b3b26..54dff4e66b 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -653,16 +653,10 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted segments = ifile.iterator(sampledPosition); while (segments.hasNext()) { - FileDataInput in = segments.next(); - try + String path = null; + try (FileDataInput in = segments.next();) { + path = in.getPath(); while (!in.isEOF()) { ByteBuffer indexKey = ByteBufferUtil.readWithShortLength(in); @@ -1495,11 +1477,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted get(SSTableReader sstable) { Descriptor desc = sstable.descriptor; @@ -2038,7 +2017,11 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted(tidy, tidy); Ref ex = lookup.putIfAbsent(desc, refc); - assert ex == null; + if (ex != null) + { + refc.close(); + throw new AssertionError(); + } return refc; } } @@ -2119,6 +2102,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted get(SSTableReader sstable) { Descriptor descriptor = sstable.descriptor; @@ -2128,7 +2112,11 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted(tidy, tidy); Ref ex = lookup.putIfAbsent(descriptor, refc); - assert ex == null; + if (ex != null) + { + refc.close(); + throw new AssertionError(); + } return refc; } } diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java index baf6d51de9..3f375e7c0c 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java @@ -177,9 +177,10 @@ public class BigTableReader extends SSTableReader Iterator segments = ifile.iterator(sampledPosition); while (segments.hasNext()) { - FileDataInput in = segments.next(); - try + String path = null; + try (FileDataInput in = segments.next()) { + path = in.getPath(); while (!in.isEOF()) { i++; @@ -220,11 +221,12 @@ public class BigTableReader extends SSTableReader if (logger.isTraceEnabled()) { // expensive sanity check! see CASSANDRA-4687 - FileDataInput fdi = dfile.getSegment(indexEntry.position); - DecoratedKey keyInDisk = partitioner.decorateKey(ByteBufferUtil.readWithShortLength(fdi)); - if (!keyInDisk.equals(key)) - throw new AssertionError(String.format("%s != %s in %s", keyInDisk, key, fdi.getPath())); - fdi.close(); + try (FileDataInput fdi = dfile.getSegment(indexEntry.position)) + { + DecoratedKey keyInDisk = partitioner.decorateKey(ByteBufferUtil.readWithShortLength(fdi)); + if (!keyInDisk.equals(key)) + throw new AssertionError(String.format("%s != %s in %s", keyInDisk, key, fdi.getPath())); + } } // store exact match for the key @@ -242,11 +244,7 @@ public class BigTableReader extends SSTableReader catch (IOException e) { markSuspect(); - throw new CorruptSSTableException(e, in.getPath()); - } - finally - { - FileUtils.closeQuietly(in); + throw new CorruptSSTableException(e, path); } } diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java index 0f8f0d3a65..30b55a00ed 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java @@ -285,6 +285,7 @@ public class BigTableWriter extends SSTableWriter return link; } + @SuppressWarnings("resource") public SSTableReader openEarly() { // find the max (exclusive) readable key @@ -318,6 +319,7 @@ public class BigTableWriter extends SSTableWriter return openFinal(makeTmpLinks(), SSTableReader.OpenReason.EARLY); } + @SuppressWarnings("resource") private SSTableReader openFinal(Descriptor desc, SSTableReader.OpenReason openReason) { if (maxDataAge < 0) @@ -507,15 +509,13 @@ public class BigTableWriter extends SSTableWriter if (components.contains(Component.FILTER)) { String path = descriptor.filenameFor(Component.FILTER); - try + try (FileOutputStream fos = new FileOutputStream(path); + DataOutputStreamPlus stream = new BufferedDataOutputStreamPlus(fos)) { // bloom filter - FileOutputStream fos = new FileOutputStream(path); - DataOutputStreamPlus stream = new BufferedDataOutputStreamPlus(fos); FilterFactory.serialize(bf, stream); stream.flush(); SyncUtil.sync(fos); - stream.close(); } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/SSTableNamesIterator.java b/src/java/org/apache/cassandra/io/sstable/format/big/SSTableNamesIterator.java index c51e59598e..7c9a3449b9 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/SSTableNamesIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/SSTableNamesIterator.java @@ -95,6 +95,7 @@ class SSTableNamesIterator extends AbstractIterator implements OnDis return fileToClose; } + @SuppressWarnings("resource") private void read(SSTableReader sstable, FileDataInput file, RowIndexEntry indexEntry) throws IOException { @@ -170,6 +171,7 @@ class SSTableNamesIterator extends AbstractIterator implements OnDis } } + @SuppressWarnings("resource") private void readIndexedColumns(CFMetaData metadata, FileDataInput file, SortedSet columnNames, diff --git a/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java b/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java index a3c4135e2b..4362cee55d 100644 --- a/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java +++ b/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java @@ -200,20 +200,14 @@ public class DataIntegrityMetadata public void writeFullChecksum(Descriptor descriptor) { File outFile = new File(descriptor.filenameFor(Component.DIGEST)); - BufferedWriter out = null; - try + try (BufferedWriter out =Files.newBufferedWriter(outFile.toPath(), Charsets.UTF_8)) { - out = Files.newBufferedWriter(outFile.toPath(), Charsets.UTF_8); out.write(String.valueOf(fullChecksum.getValue())); } catch (IOException e) { throw new FSWriteError(e, outFile); } - finally - { - FileUtils.closeQuietly(out); - } } } } diff --git a/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java b/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java index b556587a9d..6ffc8951f6 100644 --- a/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java +++ b/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java @@ -93,7 +93,7 @@ public class DataOutputBuffer extends BufferedDataOutputStreamPlus } @Override - public void close() throws IOException + public void close() { } diff --git a/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java b/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java index db9391c1a5..a5fa20bebe 100644 --- a/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java @@ -54,6 +54,7 @@ public abstract class PoolingSegmentedFile extends SegmentedFile } } + @SuppressWarnings("resource") public FileDataInput getSegment(long position) { RandomAccessReader reader = FileCacheService.instance.get(cacheKey); diff --git a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java index 302f0546b5..278f55cb5a 100644 --- a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java @@ -229,6 +229,10 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp public void deallocate() { + //make idempotent + if (buffer == null) + return; + bufferOffset += buffer.position(); FileUtils.clean(buffer); diff --git a/src/java/org/apache/cassandra/io/util/SafeMemoryWriter.java b/src/java/org/apache/cassandra/io/util/SafeMemoryWriter.java index 1096b5fa60..aad3266c10 100644 --- a/src/java/org/apache/cassandra/io/util/SafeMemoryWriter.java +++ b/src/java/org/apache/cassandra/io/util/SafeMemoryWriter.java @@ -18,7 +18,6 @@ */ package org.apache.cassandra.io.util; -import java.io.IOException; import java.nio.ByteBuffer; import java.nio.ByteOrder; @@ -26,6 +25,7 @@ public class SafeMemoryWriter extends DataOutputBuffer { private SafeMemory memory; + @SuppressWarnings("resource") public SafeMemoryWriter(long initialCapacity) { this(new SafeMemory(initialCapacity)); diff --git a/src/java/org/apache/cassandra/locator/CloudstackSnitch.java b/src/java/org/apache/cassandra/locator/CloudstackSnitch.java index afcd083b96..88c62e9710 100644 --- a/src/java/org/apache/cassandra/locator/CloudstackSnitch.java +++ b/src/java/org/apache/cassandra/locator/CloudstackSnitch.java @@ -174,15 +174,12 @@ public class CloudstackSnitch extends AbstractNetworkTopologySnitch String csEndpointFromLease(File lease) throws ConfigurationException { - BufferedReader reader = null; - - String line = null; + String line; String endpoint = null; Pattern identifierPattern = Pattern.compile("^[ \t]*option dhcp-server-identifier (.*);$"); - try + try (BufferedReader reader = new BufferedReader(new FileReader(lease))) { - reader = new BufferedReader(new FileReader(lease)); while ((line = reader.readLine()) != null) { @@ -194,14 +191,10 @@ public class CloudstackSnitch extends AbstractNetworkTopologySnitch break; } } - } + } catch (Exception e) { throw new ConfigurationException("CloudstackSnitch cannot access lease file."); - } - finally - { - FileUtils.closeQuietly(reader); } if (endpoint == null) diff --git a/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java b/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java index 247eb00864..8665816b51 100644 --- a/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java +++ b/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java @@ -136,20 +136,14 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch HashMap reloadedMap = new HashMap(); Properties properties = new Properties(); - InputStream stream = null; - try + try (InputStream stream = getClass().getClassLoader().getResourceAsStream(SNITCH_PROPERTIES_FILENAME)) { - stream = getClass().getClassLoader().getResourceAsStream(SNITCH_PROPERTIES_FILENAME); properties.load(stream); } catch (Exception e) { throw new ConfigurationException("Unable to read " + SNITCH_PROPERTIES_FILENAME, e); } - finally - { - FileUtils.closeQuietly(stream); - } for (Map.Entry entry : properties.entrySet()) { diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index c54d5eeee1..293a27cd91 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -436,6 +436,7 @@ public final class MessagingService implements MessagingServiceMBean listenGate.signalAll(); } + @SuppressWarnings("resource") private List getServerSockets(InetAddress localEp) throws ConfigurationException { final List ss = new ArrayList(2); @@ -471,6 +472,7 @@ public final class MessagingService implements MessagingServiceMBean } catch (SocketException e) { + FileUtils.closeQuietly(socket); throw new ConfigurationException("Insufficient permissions to setReuseAddress", e); } InetSocketAddress address = new InetSocketAddress(localEp, DatabaseDescriptor.getStoragePort()); @@ -480,6 +482,7 @@ public final class MessagingService implements MessagingServiceMBean } catch (BindException e) { + FileUtils.closeQuietly(socket); if (e.getMessage().contains("in use")) throw new ConfigurationException(address + " is in use by another process. Change listen_address:storage_port in cassandra.yaml to values that do not conflict with other services"); else if (e.getMessage().contains("Cannot assign requested address")) @@ -490,6 +493,7 @@ public final class MessagingService implements MessagingServiceMBean } catch (IOException e) { + FileUtils.closeQuietly(socket); throw new RuntimeException(e); } logger.info("Starting Messaging Service on port {}", DatabaseDescriptor.getStoragePort()); @@ -874,6 +878,7 @@ public final class MessagingService implements MessagingServiceMBean this.server = server; } + @SuppressWarnings("resource") public void run() { while (!server.isClosed()) diff --git a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java index 78ef61539f..0eb8e02632 100644 --- a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java +++ b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java @@ -366,6 +366,7 @@ public class OutboundTcpConnection extends Thread } } + @SuppressWarnings("resource") private boolean connect() { if (logger.isDebugEnabled()) diff --git a/src/java/org/apache/cassandra/security/SSLFactory.java b/src/java/org/apache/cassandra/security/SSLFactory.java index 956ba6749a..e9aa07d637 100644 --- a/src/java/org/apache/cassandra/security/SSLFactory.java +++ b/src/java/org/apache/cassandra/security/SSLFactory.java @@ -99,6 +99,7 @@ public final class SSLFactory return socket; } + @SuppressWarnings("resource") public static SSLContext createSSLContext(EncryptionOptions options, boolean buildTruststore) throws IOException { FileInputStream tsf = null; diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java index d350f4ef67..213edebd4d 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@ -341,6 +341,7 @@ public class ActiveRepairService * @param parentRepairSession parent repair session ID * @return Future result of all anti-compaction jobs. */ + @SuppressWarnings("resource") public ListenableFuture> doAntiCompaction(final UUID parentRepairSession, Collection> successfulRanges) { assert parentRepairSession != null; @@ -420,6 +421,7 @@ public class ActiveRepairService this.sstableMap.put(cfId, existingSSTables); } + @SuppressWarnings("resource") public synchronized Refs getAndReferenceSSTables(UUID cfId) { Set sstables = sstableMap.get(cfId); diff --git a/src/java/org/apache/cassandra/service/FileCacheService.java b/src/java/org/apache/cassandra/service/FileCacheService.java index 250e625f62..1e123101bf 100644 --- a/src/java/org/apache/cassandra/service/FileCacheService.java +++ b/src/java/org/apache/cassandra/service/FileCacheService.java @@ -143,6 +143,7 @@ public class FileCacheService } } + @SuppressWarnings("resource") public void put(CacheKey cacheKey, RandomAccessReader instance) { int memoryUsed = memoryUsage.get(); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 78376a8d31..7801c3e0a7 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -1025,8 +1025,7 @@ public class StorageProxy implements StorageProxyMBean InetAddress target = iter.next(); // Add the other destinations of the same message as a FORWARD_HEADER entry - DataOutputBuffer out = new DataOutputBuffer(); - try + try (DataOutputBuffer out = new DataOutputBuffer()) { out.writeInt(targets.size() - 1); while (iter.hasNext()) diff --git a/src/java/org/apache/cassandra/service/pager/PagingState.java b/src/java/org/apache/cassandra/service/pager/PagingState.java index ff461ab9c4..f1688800f9 100644 --- a/src/java/org/apache/cassandra/service/pager/PagingState.java +++ b/src/java/org/apache/cassandra/service/pager/PagingState.java @@ -60,9 +60,8 @@ public class PagingState public ByteBuffer serialize() { - try + try (DataOutputBuffer out = new DataOutputBufferFixed(serializedSize())) { - DataOutputBuffer out = new DataOutputBufferFixed(serializedSize()); ByteBufferUtil.writeWithShortLength(partitionKey, out); ByteBufferUtil.writeWithShortLength(cellName, out); out.writeInt(remaining); diff --git a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java index 6a33b40497..681f61ed73 100644 --- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java +++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java @@ -76,6 +76,7 @@ public class ConnectionHandler * * @throws IOException */ + @SuppressWarnings("resource") public void initiate() throws IOException { logger.debug("[Stream #{}] Sending stream init for incoming stream", session.planId()); @@ -157,6 +158,7 @@ public class ConnectionHandler protected abstract String name(); + @SuppressWarnings("resource") protected static DataOutputStreamPlus getWriteChannel(Socket socket) throws IOException { WritableByteChannel out = socket.getChannel(); @@ -175,6 +177,7 @@ public class ConnectionHandler : in; } + @SuppressWarnings("resource") public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException { StreamInitMessage message = new StreamInitMessage( @@ -246,6 +249,7 @@ public class ConnectionHandler return "STREAM-IN"; } + @SuppressWarnings("resource") public void run() { try @@ -315,6 +319,7 @@ public class ConnectionHandler messageQueue.put(message); } + @SuppressWarnings("resource") public void run() { try diff --git a/src/java/org/apache/cassandra/streaming/StreamReader.java b/src/java/org/apache/cassandra/streaming/StreamReader.java index 0f3ebb3765..1a3980d71d 100644 --- a/src/java/org/apache/cassandra/streaming/StreamReader.java +++ b/src/java/org/apache/cassandra/streaming/StreamReader.java @@ -80,6 +80,7 @@ public class StreamReader * @return SSTable transferred * @throws IOException if reading the remote sstable fails. Will throw an RTE if local write fails. */ + @SuppressWarnings("resource") public SSTableWriter read(ReadableByteChannel channel) throws IOException { logger.debug("reading file from {}, repairedAt = {}, level = {}", session.peer, repairedAt, sstableLevel); diff --git a/src/java/org/apache/cassandra/streaming/StreamWriter.java b/src/java/org/apache/cassandra/streaming/StreamWriter.java index 392dccda74..106677c13b 100644 --- a/src/java/org/apache/cassandra/streaming/StreamWriter.java +++ b/src/java/org/apache/cassandra/streaming/StreamWriter.java @@ -70,18 +70,19 @@ public class StreamWriter public void write(DataOutputStreamPlus output) throws IOException { long totalSize = totalSize(); - RandomAccessReader file = sstable.openDataReader(); - ChecksumValidator validator = new File(sstable.descriptor.filenameFor(Component.CRC)).exists() - ? DataIntegrityMetadata.checksumValidator(sstable.descriptor) - : null; - transferBuffer = validator == null ? new byte[DEFAULT_CHUNK_SIZE] : new byte[validator.chunkSize]; - // setting up data compression stream - compressedOutput = new LZFOutputStream(output); - long progress = 0L; - try + try(RandomAccessReader file = sstable.openDataReader(); + ChecksumValidator validator = new File(sstable.descriptor.filenameFor(Component.CRC)).exists() + ? DataIntegrityMetadata.checksumValidator(sstable.descriptor) + : null;) { + transferBuffer = validator == null ? new byte[DEFAULT_CHUNK_SIZE] : new byte[validator.chunkSize]; + + // setting up data compression stream + compressedOutput = new LZFOutputStream(output); + long progress = 0L; + // stream each of the required sections of the file for (Pair section : sections) { @@ -109,12 +110,6 @@ public class StreamWriter compressedOutput.flush(); } } - finally - { - // no matter what happens close file - FileUtils.closeQuietly(file); - FileUtils.closeQuietly(validator); - } } protected long totalSize() diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java index 89773ead25..1936a94fc1 100644 --- a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java +++ b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java @@ -61,6 +61,7 @@ public class CompressedStreamReader extends StreamReader * @throws java.io.IOException if reading the remote sstable fails. Will throw an RTE if local write fails. */ @Override + @SuppressWarnings("resource") public SSTableWriter read(ReadableByteChannel channel) throws IOException { logger.debug("reading file from {}, repairedAt = {}", session.peer, repairedAt); diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java index 2fd7f6351d..144980cecb 100644 --- a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java +++ b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java @@ -55,14 +55,11 @@ public class CompressedStreamWriter extends StreamWriter public void write(DataOutputStreamPlus out) throws IOException { long totalSize = totalSize(); - RandomAccessReader file = sstable.openDataReader(); - final ChannelProxy fc = file.getChannel(); - - long progress = 0L; - // calculate chunks to transfer. we want to send continuous chunks altogether. - List> sections = getTransferSections(compressionInfo.chunks); - try + try (RandomAccessReader file = sstable.openDataReader(); final ChannelProxy fc = file.getChannel()) { + long progress = 0L; + // calculate chunks to transfer. we want to send continuous chunks altogether. + List> sections = getTransferSections(compressionInfo.chunks); // stream each of the required sections of the file for (final Pair section : sections) { @@ -75,7 +72,7 @@ public class CompressedStreamWriter extends StreamWriter final long bytesTransferredFinal = bytesTransferred; final int toTransfer = (int) Math.min(CHUNK_SIZE, length - bytesTransferred); limiter.acquire(toTransfer); - long lastWrite = out.applyToChannel( new Function() + long lastWrite = out.applyToChannel(new Function() { public Long apply(WritableByteChannel wbc) { @@ -88,11 +85,6 @@ public class CompressedStreamWriter extends StreamWriter } } } - finally - { - // no matter what happens close file - FileUtils.closeQuietly(file); - } } @Override diff --git a/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java b/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java index 14186511c6..fdfb32e05f 100644 --- a/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java +++ b/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java @@ -36,6 +36,7 @@ public class IncomingFileMessage extends StreamMessage { public static Serializer serializer = new Serializer() { + @SuppressWarnings("resource") public IncomingFileMessage deserialize(ReadableByteChannel in, int version, StreamSession session) throws IOException { DataInputStream input = new DataInputStream(Channels.newInputStream(in)); diff --git a/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java b/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java index 4928039feb..e8b3f828b3 100644 --- a/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java +++ b/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java @@ -83,9 +83,11 @@ public class StreamInitMessage try { int size = (int)StreamInitMessage.serializer.serializedSize(this, version); - DataOutputBuffer buffer = new DataOutputBufferFixed(size); - StreamInitMessage.serializer.serialize(this, buffer, version); - bytes = buffer.getData(); + try (DataOutputBuffer buffer = new DataOutputBufferFixed(size)) + { + StreamInitMessage.serializer.serialize(this, buffer, version); + bytes = buffer.getData(); + } } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/thrift/CustomTNonBlockingServer.java b/src/java/org/apache/cassandra/thrift/CustomTNonBlockingServer.java index 11fcc5e6be..de8df57232 100644 --- a/src/java/org/apache/cassandra/thrift/CustomTNonBlockingServer.java +++ b/src/java/org/apache/cassandra/thrift/CustomTNonBlockingServer.java @@ -37,6 +37,7 @@ public class CustomTNonBlockingServer extends TNonblockingServer } @Override + @SuppressWarnings("resource") protected boolean requestInvoke(FrameBuffer frameBuffer) { TNonblockingSocket socket = (TNonblockingSocket)((CustomFrameBuffer)frameBuffer).getTransport(); @@ -47,6 +48,7 @@ public class CustomTNonBlockingServer extends TNonblockingServer public static class Factory implements TServerFactory { + @SuppressWarnings("resource") public TServer buildTServer(Args args) { if (DatabaseDescriptor.getClientEncryptionOptions().enabled) diff --git a/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java b/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java index e7584c9383..a025004172 100644 --- a/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java +++ b/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java @@ -84,6 +84,7 @@ public class CustomTThreadPoolServer extends TServer this.args = args; } + @SuppressWarnings("resource") public void serve() { try @@ -184,18 +185,16 @@ public class CustomTThreadPoolServer extends TServer public void run() { TProcessor processor = null; - TTransport inputTransport = null; - TTransport outputTransport = null; TProtocol inputProtocol = null; TProtocol outputProtocol = null; SocketAddress socket = null; - try + try (TTransport inputTransport = inputTransportFactory_.getTransport(client_); + TTransport outputTransport = outputTransportFactory_.getTransport(client_)) { socket = ((TCustomSocket) client_).getSocket().getRemoteSocketAddress(); ThriftSessionManager.instance.setCurrentSocket(socket); processor = processorFactory_.getProcessor(client_); - inputTransport = inputTransportFactory_.getTransport(client_); - outputTransport = outputTransportFactory_.getTransport(client_); + inputProtocol = inputProtocolFactory_.getProtocol(inputTransport); outputProtocol = outputProtocolFactory_.getProtocol(outputTransport); // we check stopped first to make sure we're not supposed to be shutting @@ -227,10 +226,7 @@ public class CustomTThreadPoolServer extends TServer { if (socket != null) ThriftSessionManager.instance.connectionComplete(socket); - if (inputTransport != null) - inputTransport.close(); - if (outputTransport != null) - outputTransport.close(); + activeClients.decrementAndGet(); } } @@ -238,6 +234,7 @@ public class CustomTThreadPoolServer extends TServer public static class Factory implements TServerFactory { + @SuppressWarnings("resource") public TServer buildTServer(Args args) { final InetSocketAddress addr = args.addr; diff --git a/src/java/org/apache/cassandra/thrift/SSLTransportFactory.java b/src/java/org/apache/cassandra/thrift/SSLTransportFactory.java index d80d76e1df..ea74b94f93 100644 --- a/src/java/org/apache/cassandra/thrift/SSLTransportFactory.java +++ b/src/java/org/apache/cassandra/thrift/SSLTransportFactory.java @@ -52,6 +52,7 @@ public class SSLTransportFactory implements ITransportFactory private String[] cipherSuites; @Override + @SuppressWarnings("resource") public TTransport openTransport(String host, int port) throws Exception { TSSLTransportFactory.TSSLTransportParameters params = new TSSLTransportFactory.TSSLTransportParameters(protocol, cipherSuites); diff --git a/src/java/org/apache/cassandra/thrift/TCustomNonblockingServerSocket.java b/src/java/org/apache/cassandra/thrift/TCustomNonblockingServerSocket.java index 63466b80bf..a430721828 100644 --- a/src/java/org/apache/cassandra/thrift/TCustomNonblockingServerSocket.java +++ b/src/java/org/apache/cassandra/thrift/TCustomNonblockingServerSocket.java @@ -43,6 +43,7 @@ public class TCustomNonblockingServerSocket extends TNonblockingServerSocket } @Override + @SuppressWarnings("resource") protected TNonblockingSocket acceptImpl() throws TTransportException { TNonblockingSocket tsocket = super.acceptImpl(); diff --git a/src/java/org/apache/cassandra/thrift/TCustomServerSocket.java b/src/java/org/apache/cassandra/thrift/TCustomServerSocket.java index 477ef8c004..8e27481c3a 100644 --- a/src/java/org/apache/cassandra/thrift/TCustomServerSocket.java +++ b/src/java/org/apache/cassandra/thrift/TCustomServerSocket.java @@ -89,6 +89,7 @@ public class TCustomServerSocket extends TServerTransport } @Override + @SuppressWarnings("resource") protected TCustomSocket acceptImpl() throws TTransportException { diff --git a/src/java/org/apache/cassandra/thrift/TFramedTransportFactory.java b/src/java/org/apache/cassandra/thrift/TFramedTransportFactory.java index a4c6bb772e..7bf0b96044 100644 --- a/src/java/org/apache/cassandra/thrift/TFramedTransportFactory.java +++ b/src/java/org/apache/cassandra/thrift/TFramedTransportFactory.java @@ -34,6 +34,7 @@ public class TFramedTransportFactory implements ITransportFactory private static final String THRIFT_FRAMED_TRANSPORT_SIZE_IN_MB = "cassandra.thrift.framed.size_mb"; private int thriftFramedTransportSizeMb = 15; // 15Mb is the default for C* & Hadoop ConfigHelper + @SuppressWarnings("resource") public TTransport openTransport(String host, int port) throws TTransportException { TSocket socket = new TSocket(host, port); diff --git a/src/java/org/apache/cassandra/thrift/THsHaDisruptorServer.java b/src/java/org/apache/cassandra/thrift/THsHaDisruptorServer.java index dd501ec5e7..37bc4409cf 100644 --- a/src/java/org/apache/cassandra/thrift/THsHaDisruptorServer.java +++ b/src/java/org/apache/cassandra/thrift/THsHaDisruptorServer.java @@ -67,6 +67,7 @@ public class THsHaDisruptorServer extends TDisruptorServer public static class Factory implements TServerFactory { + @SuppressWarnings("resource") public TServer buildTServer(Args args) { if (DatabaseDescriptor.getClientEncryptionOptions().enabled) diff --git a/src/java/org/apache/cassandra/tools/SSTableExport.java b/src/java/org/apache/cassandra/tools/SSTableExport.java index 61edad200b..bc460a1792 100644 --- a/src/java/org/apache/cassandra/tools/SSTableExport.java +++ b/src/java/org/apache/cassandra/tools/SSTableExport.java @@ -225,8 +225,7 @@ public class SSTableExport public static void enumeratekeys(Descriptor desc, PrintStream outs, CFMetaData metadata) throws IOException { - KeyIterator iter = new KeyIterator(desc); - try + try (KeyIterator iter = new KeyIterator(desc)) { DecoratedKey lastKey = null; while (iter.hasNext()) @@ -242,10 +241,6 @@ public class SSTableExport checkStream(outs); // flushes } } - finally - { - iter.close(); - } } /** @@ -261,8 +256,8 @@ public class SSTableExport public static void export(Descriptor desc, PrintStream outs, Collection toExport, String[] excludes, CFMetaData metadata) throws IOException { SSTableReader sstable = SSTableReader.open(desc); - RandomAccessReader dfile = sstable.openDataReader(); - try + + try (RandomAccessReader dfile = sstable.openDataReader()) { IPartitioner partitioner = sstable.partitioner; @@ -305,10 +300,6 @@ public class SSTableExport outs.println("\n]"); outs.flush(); } - finally - { - dfile.close(); - } } // This is necessary to accommodate the test suite since you cannot open a Reader more diff --git a/src/java/org/apache/cassandra/tools/SSTableImport.java b/src/java/org/apache/cassandra/tools/SSTableImport.java index 7b187acbe8..2fda6bd6ae 100644 --- a/src/java/org/apache/cassandra/tools/SSTableImport.java +++ b/src/java/org/apache/cassandra/tools/SSTableImport.java @@ -299,54 +299,58 @@ public class SSTableImport int importedKeys = 0; long start = System.nanoTime(); - JsonParser parser = getParser(jsonFile); - - Object[] data = parser.readValueAs(new TypeReference(){}); + Object[] data; + try (JsonParser parser = getParser(jsonFile)) + { + data = parser.readValueAs(new TypeReference(){}); + } keyCountToImport = (keyCountToImport == null) ? data.length : keyCountToImport; - SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(ssTablePath), keyCountToImport, ActiveRepairService.UNREPAIRED_SSTABLE, 0); - System.out.printf("Importing %s keys...%n", keyCountToImport); - - // sort by dk representation, but hold onto the hex version - SortedMap> decoratedKeys = new TreeMap>(); - - for (Object row : data) + try (SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(ssTablePath), keyCountToImport, ActiveRepairService.UNREPAIRED_SSTABLE, 0)) { - Map rowAsMap = (Map)row; - decoratedKeys.put(partitioner.decorateKey(getKeyValidator(columnFamily).fromString((String) rowAsMap.get("key"))), rowAsMap); - } + System.out.printf("Importing %s keys...%n", keyCountToImport); - for (Map.Entry> row : decoratedKeys.entrySet()) - { - if (row.getValue().containsKey("metadata")) + // sort by dk representation, but hold onto the hex version + SortedMap> decoratedKeys = new TreeMap>(); + + for (Object row : data) { - parseMeta((Map) row.getValue().get("metadata"), columnFamily, null); + Map rowAsMap = (Map) row; + decoratedKeys.put(partitioner.decorateKey(getKeyValidator(columnFamily).fromString((String) rowAsMap.get("key"))), rowAsMap); } - Object columns = row.getValue().get("cells"); - addColumnsToCF((List) columns, columnFamily); - - - writer.append(row.getKey(), columnFamily); - columnFamily.clear(); - - importedKeys++; - - long current = System.nanoTime(); - - if (TimeUnit.NANOSECONDS.toSeconds(current - start) >= 5) // 5 secs. + for (Map.Entry> row : decoratedKeys.entrySet()) { - System.out.printf("Currently imported %d keys.%n", importedKeys); - start = current; + if (row.getValue().containsKey("metadata")) + { + parseMeta((Map) row.getValue().get("metadata"), columnFamily, null); + } + + Object columns = row.getValue().get("cells"); + addColumnsToCF((List) columns, columnFamily); + + + writer.append(row.getKey(), columnFamily); + columnFamily.clear(); + + importedKeys++; + + long current = System.nanoTime(); + + if (TimeUnit.NANOSECONDS.toSeconds(current - start) >= 5) // 5 secs. + { + System.out.printf("Currently imported %d keys.%n", importedKeys); + start = current; + } + + if (keyCountToImport == importedKeys) + break; } - if (keyCountToImport == importedKeys) - break; + writer.finish(true); } - writer.finish(true); - return importedKeys; } @@ -356,28 +360,29 @@ public class SSTableImport int importedKeys = 0; // already imported keys count long start = System.nanoTime(); - JsonParser parser = getParser(jsonFile); - - if (keyCountToImport == null) + try (JsonParser parser = getParser(jsonFile)) { - keyCountToImport = 0; - System.out.println("Counting keys to import, please wait... (NOTE: to skip this use -n )"); - parser.nextToken(); // START_ARRAY - while (parser.nextToken() != null) + if (keyCountToImport == null) { - parser.skipChildren(); - if (parser.getCurrentToken() == JsonToken.END_ARRAY) - break; + keyCountToImport = 0; + System.out.println("Counting keys to import, please wait... (NOTE: to skip this use -n )"); - keyCountToImport++; + parser.nextToken(); // START_ARRAY + while (parser.nextToken() != null) + { + parser.skipChildren(); + if (parser.getCurrentToken() == JsonToken.END_ARRAY) + break; + + keyCountToImport++; + } } + System.out.printf("Importing %s keys...%n", keyCountToImport); } - System.out.printf("Importing %s keys...%n", keyCountToImport); - - parser = getParser(jsonFile); // renewing parser - try (SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(ssTablePath), keyCountToImport, ActiveRepairService.UNREPAIRED_SSTABLE);) + try (JsonParser parser = getParser(jsonFile); // renewing parser + SSTableWriter writer = SSTableWriter.create(Descriptor.fromFilename(ssTablePath), keyCountToImport, ActiveRepairService.UNREPAIRED_SSTABLE);) { int lineNumber = 1; DecoratedKey prevStoredKey = null; diff --git a/src/java/org/apache/cassandra/tools/StandaloneScrubber.java b/src/java/org/apache/cassandra/tools/StandaloneScrubber.java index d32ef8805d..dd513b8503 100644 --- a/src/java/org/apache/cassandra/tools/StandaloneScrubber.java +++ b/src/java/org/apache/cassandra/tools/StandaloneScrubber.java @@ -119,8 +119,7 @@ public class StandaloneScrubber { try (LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.SCRUB, sstable)) { - Scrubber scrubber = new Scrubber(cfs, txn, options.skipCorrupted, handler, true, !options.noValidate); - try + try (Scrubber scrubber = new Scrubber(cfs, txn, options.skipCorrupted, handler, true, !options.noValidate)) { scrubber.scrub(); } @@ -132,10 +131,6 @@ public class StandaloneScrubber throw t; } } - finally - { - scrubber.close(); - } // Remove the sstable (it's been copied by scrub and snapshotted) sstable.markObsolete(); diff --git a/src/java/org/apache/cassandra/tools/StandaloneVerifier.java b/src/java/org/apache/cassandra/tools/StandaloneVerifier.java index a4f3e80355..f71f58df62 100644 --- a/src/java/org/apache/cassandra/tools/StandaloneVerifier.java +++ b/src/java/org/apache/cassandra/tools/StandaloneVerifier.java @@ -100,8 +100,8 @@ public class StandaloneVerifier { try { - Verifier verifier = new Verifier(cfs, sstable, handler, true); - try + + try (Verifier verifier = new Verifier(cfs, sstable, handler, true)) { verifier.verify(extended); } @@ -110,10 +110,6 @@ public class StandaloneVerifier System.err.println(String.format("Error verifying %s: %s", sstable, cs.getMessage())); hasFailed = true; } - finally - { - verifier.close(); - } } catch (Exception e) { diff --git a/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java b/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java index 5fad3ea1aa..6f80ac0ad5 100644 --- a/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java +++ b/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java @@ -40,6 +40,7 @@ class BloomFilterSerializer implements ISerializer return deserialize(in, false); } + @SuppressWarnings("resource") public BloomFilter deserialize(DataInput in, boolean offheap) throws IOException { int hashes = in.readInt(); diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index c20e33eabc..17edeb0d10 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -347,10 +347,8 @@ public class FBUtilities public static String getReleaseVersionString() { - InputStream in = null; - try + try (InputStream in = FBUtilities.class.getClassLoader().getResourceAsStream("org/apache/cassandra/config/version.properties")) { - in = FBUtilities.class.getClassLoader().getResourceAsStream("org/apache/cassandra/config/version.properties"); if (in == null) { return System.getProperty("cassandra.releaseVersion", "Unknown"); @@ -365,10 +363,6 @@ public class FBUtilities logger.warn("Unable to load version.properties", e); return "debug version"; } - finally - { - FileUtils.closeQuietly(in); - } } public static long timestampMicros() @@ -718,10 +712,10 @@ public class FBUtilities public static byte[] serialize(T object, IVersionedSerializer serializer, int version) { - try + int size = (int) serializer.serializedSize(object, version); + + try (DataOutputBuffer buffer = new DataOutputBufferFixed(size)) { - int size = (int) serializer.serializedSize(object, version); - DataOutputBuffer buffer = new DataOutputBufferFixed(size); serializer.serialize(object, buffer, version); assert buffer.getLength() == size && buffer.getData().length == size : String.format("Final buffer length %s to accommodate data size of %s (predicted %s) for %s", diff --git a/src/java/org/apache/cassandra/utils/FilterFactory.java b/src/java/org/apache/cassandra/utils/FilterFactory.java index 7cfc3327bf..d77500c32f 100644 --- a/src/java/org/apache/cassandra/utils/FilterFactory.java +++ b/src/java/org/apache/cassandra/utils/FilterFactory.java @@ -78,6 +78,7 @@ public class FilterFactory return createFilter(spec.K, numElements, spec.bucketsPerElement, offheap); } + @SuppressWarnings("resource") private static IFilter createFilter(int hash, long numElements, int bucketsPer, boolean offheap) { long numBits = (numElements * bucketsPer) + BITSET_EXCESS; diff --git a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java index 5b46700699..5448390ad4 100644 --- a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java +++ b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java @@ -61,9 +61,9 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client if (username != null && password != null) builder = builder.withCredentials(username, password); - try (Cluster cluster = builder.build()) + try (Cluster cluster = builder.build(); Session session = cluster.connect()) { - Session session = cluster.connect(); + Metadata metadata = cluster.getMetadata(); setPartitioner(metadata.getPartitioner()); diff --git a/src/java/org/apache/cassandra/utils/concurrent/Ref.java b/src/java/org/apache/cassandra/utils/concurrent/Ref.java index ebabd79cc5..3df3ea452b 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Ref.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Ref.java @@ -49,7 +49,7 @@ import static org.apache.cassandra.utils.Throwables.merge; * Once the Ref.GlobalState has been completely released, the Tidy method is called and it removes the global reference * to itself so it may also be collected. */ -public final class Ref implements RefCounted, AutoCloseable +public final class Ref implements RefCounted { static final Logger logger = LoggerFactory.getLogger(Ref.class); public static final boolean DEBUG_ENABLED = System.getProperty("cassandra.debugrefcount", "false").equalsIgnoreCase("true"); diff --git a/src/java/org/apache/cassandra/utils/concurrent/Refs.java b/src/java/org/apache/cassandra/utils/concurrent/Refs.java index dd6597160f..bbc672d645 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Refs.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Refs.java @@ -89,6 +89,12 @@ public final class Refs> extends AbstractCollection i return ref != null; } + public void relaseAllExcept(Collection keep) + { + Collection release = new ArrayList<>(references.keySet()); + release.retainAll(keep); + release(release); + } /** * Release a retained Ref to all of the provided objects; if any is not held, an exception will be thrown * @param release diff --git a/src/java/org/apache/cassandra/utils/obs/OffHeapBitSet.java b/src/java/org/apache/cassandra/utils/obs/OffHeapBitSet.java index 02efa65758..46c1bd04e7 100644 --- a/src/java/org/apache/cassandra/utils/obs/OffHeapBitSet.java +++ b/src/java/org/apache/cassandra/utils/obs/OffHeapBitSet.java @@ -125,6 +125,7 @@ public class OffHeapBitSet implements IBitSet return type.sizeof((int) bytes.size()) + bytes.size(); } + @SuppressWarnings("resource") public static OffHeapBitSet deserialize(DataInput in) throws IOException { long byteCount = in.readInt() * 8L;