From 928dc9d7f653762879efd156bf3049f0f5a7b48f Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Thu, 30 Jul 2026 23:26:10 +0800 Subject: [PATCH 1/4] Fix CodeQL resource lifetime warnings --- .../aliyun/OssDataSegmentPullerTest.java | 123 +++++++------ .../druid/storage/aliyun/OssTaskLogsTest.java | 36 ++-- .../movingaverage/MovingAverageQueryTest.java | 18 +- .../TimestampGroupByAggregationTest.java | 38 ++-- .../druid/query/filter/BloomKFilterTest.java | 77 ++++---- .../storage/s3/S3DataSegmentPullerTest.java | 9 +- .../druid/storage/s3/S3TaskLogsTest.java | 45 +++-- .../batch/parallel/HttpShuffleClientTest.java | 16 +- .../SQLMetadataStorageActionHandlerTest.java | 9 +- .../util/common/io/smoosh/FileSmoosher.java | 4 +- .../segment/file/SegmentFileBuilderV10.java | 2 + .../input/impl/RetryingInputStreamTest.java | 30 +-- .../util/common/CompressionUtilsTest.java | 174 +++++++++++++----- .../aggregation/AggregationTestHelper.java | 49 ++--- .../druid/metadata/input/SqlEntityTest.java | 8 +- .../apache/druid/rpc/RequestBuilderTest.java | 20 +- .../druid/server/QueryResourceTest.java | 30 ++- .../server/initialization/JettyTest.java | 34 ++-- .../sql/avatica/DruidAvaticaHandlerTest.java | 117 ++++++------ 19 files changed, 503 insertions(+), 336 deletions(-) diff --git a/extensions-contrib/aliyun-oss-extensions/src/test/java/org/apache/druid/storage/aliyun/OssDataSegmentPullerTest.java b/extensions-contrib/aliyun-oss-extensions/src/test/java/org/apache/druid/storage/aliyun/OssDataSegmentPullerTest.java index e3d667a411d6..0fb2ec9c7c3b 100644 --- a/extensions-contrib/aliyun-oss-extensions/src/test/java/org/apache/druid/storage/aliyun/OssDataSegmentPullerTest.java +++ b/extensions-contrib/aliyun-oss-extensions/src/test/java/org/apache/druid/storage/aliyun/OssDataSegmentPullerTest.java @@ -38,6 +38,7 @@ import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; +import java.io.InputStream; import java.io.OutputStream; import java.net.URI; import java.nio.charset.StandardCharsets; @@ -95,7 +96,8 @@ public void testGZUncompress() throws IOException, SegmentLoadingException final File tmpFile = temporaryFolder.newFile("gzTest.gz"); - try (OutputStream outputStream = new GZIPOutputStream(new FileOutputStream(tmpFile))) { + try (final FileOutputStream fileOutputStream = new FileOutputStream(tmpFile); + final OutputStream outputStream = new GZIPOutputStream(fileOutputStream)) { outputStream.write(value); } @@ -103,7 +105,6 @@ public void testGZUncompress() throws IOException, SegmentLoadingException object0.setBucketName(bucket); object0.setKey(keyPrefix + "/renames-0.gz"); object0.getObjectMetadata().setLastModified(new Date(0)); - object0.setObjectContent(new FileInputStream(tmpFile)); final OSSObjectSummary objectSummary = new OSSObjectSummary(); objectSummary.setBucketName(bucket); @@ -115,30 +116,33 @@ public void testGZUncompress() throws IOException, SegmentLoadingException final File tmpDir = temporaryFolder.newFolder("gzTestDir"); - EasyMock.expect(ossClient.doesObjectExist(EasyMock.eq(object0.getBucketName()), EasyMock.eq(object0.getKey()))) - .andReturn(true) - .once(); - EasyMock.expect(ossClient.getObjectMetadata(object0.getBucketName(), object0.getKey())) - .andReturn(objectMetadata) - .once(); - EasyMock.expect(ossClient.getObject(EasyMock.eq(object0.getBucketName()), EasyMock.eq(object0.getKey()))) - .andReturn(object0) - .once(); - OssDataSegmentPuller puller = new OssDataSegmentPuller(ossClient); - - EasyMock.replay(ossClient); - FileUtils.FileCopyResult result = puller.getSegmentFiles( - new CloudObjectLocation( - bucket, - object0.getKey() - ), tmpDir - ); - EasyMock.verify(ossClient); - - Assert.assertEquals(value.length, result.size()); - File expected = new File(tmpDir, "renames-0"); - Assert.assertTrue(expected.exists()); - Assert.assertEquals(value.length, expected.length()); + try (final InputStream objectContent = new FileInputStream(tmpFile)) { + object0.setObjectContent(objectContent); + EasyMock.expect(ossClient.doesObjectExist(EasyMock.eq(object0.getBucketName()), EasyMock.eq(object0.getKey()))) + .andReturn(true) + .once(); + EasyMock.expect(ossClient.getObjectMetadata(object0.getBucketName(), object0.getKey())) + .andReturn(objectMetadata) + .once(); + EasyMock.expect(ossClient.getObject(EasyMock.eq(object0.getBucketName()), EasyMock.eq(object0.getKey()))) + .andReturn(object0) + .once(); + final OssDataSegmentPuller puller = new OssDataSegmentPuller(ossClient); + + EasyMock.replay(ossClient); + final FileUtils.FileCopyResult result = puller.getSegmentFiles( + new CloudObjectLocation( + bucket, + object0.getKey() + ), tmpDir + ); + EasyMock.verify(ossClient); + + Assert.assertEquals(value.length, result.size()); + final File expected = new File(tmpDir, "renames-0"); + Assert.assertTrue(expected.exists()); + Assert.assertEquals(value.length, expected.length()); + } } @Test @@ -151,7 +155,8 @@ public void testGZUncompressRetries() throws IOException, SegmentLoadingExceptio final File tmpFile = temporaryFolder.newFile("gzTest.gz"); - try (OutputStream outputStream = new GZIPOutputStream(new FileOutputStream(tmpFile))) { + try (final FileOutputStream fileOutputStream = new FileOutputStream(tmpFile); + final OutputStream outputStream = new GZIPOutputStream(fileOutputStream)) { outputStream.write(value); } @@ -160,7 +165,6 @@ public void testGZUncompressRetries() throws IOException, SegmentLoadingExceptio object0.setBucketName(bucket); object0.setKey(keyPrefix + "/renames-0.gz"); object0.getObjectMetadata().setLastModified(new Date(0)); - object0.setObjectContent(new FileInputStream(tmpFile)); final ObjectMetadata objectMetadata = new ObjectMetadata(); objectMetadata.setLastModified(new Date(0)); @@ -168,36 +172,39 @@ public void testGZUncompressRetries() throws IOException, SegmentLoadingExceptio File tmpDir = temporaryFolder.newFolder("gzTestDir"); OSSException exception = new OSSException("OssDataSegmentPullerTest", "NoSuchKey", null, null, null, null, null); - EasyMock.expect(ossClient.doesObjectExist(EasyMock.eq(object0.getBucketName()), EasyMock.eq(object0.getKey()))) - .andReturn(true) - .once(); - EasyMock.expect(ossClient.getObjectMetadata(bucket, object0.getKey())) - .andReturn(objectMetadata) - .once(); - EasyMock.expect(ossClient.getObject(EasyMock.eq(bucket), EasyMock.eq(object0.getKey()))) - .andThrow(exception) - .once(); - EasyMock.expect(ossClient.getObjectMetadata(bucket, object0.getKey())) - .andReturn(objectMetadata) - .once(); - EasyMock.expect(ossClient.getObject(EasyMock.eq(bucket), EasyMock.eq(object0.getKey()))) - .andReturn(object0) - .once(); - OssDataSegmentPuller puller = new OssDataSegmentPuller(ossClient); - - EasyMock.replay(ossClient); - FileUtils.FileCopyResult result = puller.getSegmentFiles( - new CloudObjectLocation( - bucket, - object0.getKey() - ), tmpDir - ); - EasyMock.verify(ossClient); - - Assert.assertEquals(value.length, result.size()); - File expected = new File(tmpDir, "renames-0"); - Assert.assertTrue(expected.exists()); - Assert.assertEquals(value.length, expected.length()); + try (final InputStream objectContent = new FileInputStream(tmpFile)) { + object0.setObjectContent(objectContent); + EasyMock.expect(ossClient.doesObjectExist(EasyMock.eq(object0.getBucketName()), EasyMock.eq(object0.getKey()))) + .andReturn(true) + .once(); + EasyMock.expect(ossClient.getObjectMetadata(bucket, object0.getKey())) + .andReturn(objectMetadata) + .once(); + EasyMock.expect(ossClient.getObject(EasyMock.eq(bucket), EasyMock.eq(object0.getKey()))) + .andThrow(exception) + .once(); + EasyMock.expect(ossClient.getObjectMetadata(bucket, object0.getKey())) + .andReturn(objectMetadata) + .once(); + EasyMock.expect(ossClient.getObject(EasyMock.eq(bucket), EasyMock.eq(object0.getKey()))) + .andReturn(object0) + .once(); + final OssDataSegmentPuller puller = new OssDataSegmentPuller(ossClient); + + EasyMock.replay(ossClient); + final FileUtils.FileCopyResult result = puller.getSegmentFiles( + new CloudObjectLocation( + bucket, + object0.getKey() + ), tmpDir + ); + EasyMock.verify(ossClient); + + Assert.assertEquals(value.length, result.size()); + final File expected = new File(tmpDir, "renames-0"); + Assert.assertTrue(expected.exists()); + Assert.assertEquals(value.length, expected.length()); + } } } diff --git a/extensions-contrib/aliyun-oss-extensions/src/test/java/org/apache/druid/storage/aliyun/OssTaskLogsTest.java b/extensions-contrib/aliyun-oss-extensions/src/test/java/org/apache/druid/storage/aliyun/OssTaskLogsTest.java index 16b09866ec4b..d88aa7af4680 100644 --- a/extensions-contrib/aliyun-oss-extensions/src/test/java/org/apache/druid/storage/aliyun/OssTaskLogsTest.java +++ b/extensions-contrib/aliyun-oss-extensions/src/test/java/org/apache/druid/storage/aliyun/OssTaskLogsTest.java @@ -300,10 +300,11 @@ public void test_taskLog_fetch() throws IOException OssTaskLogs ossTaskLogs = getOssTaskLogs(); Optional inputStreamOptional = ossTaskLogs.streamTaskLog(KEY_1, 0); - String taskLogs = new BufferedReader( - new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) - .lines() - .collect(Collectors.joining("\n")); + final String taskLogs; + try (final BufferedReader reader = new BufferedReader( + new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8))) { + taskLogs = reader.lines().collect(Collectors.joining("\n")); + } Assert.assertEquals(LOG_CONTENTS, taskLogs); } @@ -324,10 +325,11 @@ public void test_taskLog_fetch_withRange() throws IOException OssTaskLogs ossTaskLogs = getOssTaskLogs(); Optional inputStreamOptional = ossTaskLogs.streamTaskLog(KEY_1, 1); - String taskLogs = new BufferedReader( - new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) - .lines() - .collect(Collectors.joining("\n")); + final String taskLogs; + try (final BufferedReader reader = new BufferedReader( + new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8))) { + taskLogs = reader.lines().collect(Collectors.joining("\n")); + } Assert.assertEquals(LOG_CONTENTS.substring(1), taskLogs); } @@ -348,10 +350,11 @@ public void test_taskLog_fetch_withNegativeRange() throws IOException OssTaskLogs ossTaskLogs = getOssTaskLogs(); Optional inputStreamOptional = ossTaskLogs.streamTaskLog(KEY_1, -1 * (LOG_CONTENTS.length() - 1)); - String taskLogs = new BufferedReader( - new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) - .lines() - .collect(Collectors.joining("\n")); + final String taskLogs; + try (final BufferedReader reader = new BufferedReader( + new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8))) { + taskLogs = reader.lines().collect(Collectors.joining("\n")); + } Assert.assertEquals(LOG_CONTENTS.substring(1), taskLogs); } @@ -373,10 +376,11 @@ public void test_taskReport_fetch() throws IOException OssTaskLogs ossTaskLogs = getOssTaskLogs(); Optional inputStreamOptional = ossTaskLogs.streamTaskReports(KEY_1); - String report = new BufferedReader( - new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) - .lines() - .collect(Collectors.joining("\n")); + final String report; + try (final BufferedReader reader = new BufferedReader( + new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8))) { + report = reader.lines().collect(Collectors.joining("\n")); + } Assert.assertEquals(REPORT_CONTENTS, report); } diff --git a/extensions-contrib/moving-average-query/src/test/java/org/apache/druid/query/movingaverage/MovingAverageQueryTest.java b/extensions-contrib/moving-average-query/src/test/java/org/apache/druid/query/movingaverage/MovingAverageQueryTest.java index d16895dc7712..6812fdf4f403 100644 --- a/extensions-contrib/moving-average-query/src/test/java/org/apache/druid/query/movingaverage/MovingAverageQueryTest.java +++ b/extensions-contrib/moving-average-query/src/test/java/org/apache/druid/query/movingaverage/MovingAverageQueryTest.java @@ -117,12 +117,17 @@ public class MovingAverageQueryTest extends InitializedNullHandlingTest @Parameters(name = "{0}") public static Iterable data() throws IOException { - BufferedReader testReader = new BufferedReader( - new InputStreamReader(MovingAverageQueryTest.class.getResourceAsStream("/queryTests"), StandardCharsets.UTF_8)); List tests = new ArrayList<>(); - for (String line = testReader.readLine(); line != null; line = testReader.readLine()) { - tests.add(new String[]{line}); + try (final BufferedReader testReader = new BufferedReader( + new InputStreamReader( + MovingAverageQueryTest.class.getResourceAsStream("/queryTests"), + StandardCharsets.UTF_8 + ) + )) { + for (String line = testReader.readLine(); line != null; line = testReader.readLine()) { + tests.add(new String[]{line}); + } } return tests; @@ -171,9 +176,10 @@ public QueryRunner getQueryRunnerForSegments(Query query, Iterable seq = helper.createIndexAndRunQueryOnSegment( - zip.getInputStream(zip.getEntry("druid.sample.tsv")), - new InputRowSchema( - new TimestampSpec("timestamp", "auto", null), - new DimensionsSpec(DimensionsSpec.getDefaultSchemas(List.of("product"))), - ColumnsFilter.all() - ), - DelimitedInputFormat.forColumns( - List.of("timestamp", "cat", "product", "prefer", "prefer2", "pty_country") - ), - aggregators, - 0, - Granularities.MONTH, - 100, - groupByQuery + final Sequence seq; + try (final ZipFile zip = new ZipFile( + new File(this.getClass().getClassLoader().getResource("druid.sample.tsv.zip").toURI()) ); + final InputStream inputStream = zip.getInputStream(zip.getEntry("druid.sample.tsv"))) { + seq = helper.createIndexAndRunQueryOnSegment( + inputStream, + new InputRowSchema( + new TimestampSpec("timestamp", "auto", null), + new DimensionsSpec(DimensionsSpec.getDefaultSchemas(List.of("product"))), + ColumnsFilter.all() + ), + DelimitedInputFormat.forColumns( + List.of("timestamp", "cat", "product", "prefer", "prefer2", "pty_country") + ), + aggregators, + 0, + Granularities.MONTH, + 100, + groupByQuery + ); + } int groupByFieldNumber = groupByQuery.getResultRowSignature().indexOf(groupByField); diff --git a/extensions-core/druid-bloom-filter/src/test/java/org/apache/druid/query/filter/BloomKFilterTest.java b/extensions-core/druid-bloom-filter/src/test/java/org/apache/druid/query/filter/BloomKFilterTest.java index 63bd658cfc77..bc87c2c2f5c5 100644 --- a/extensions-core/druid-bloom-filter/src/test/java/org/apache/druid/query/filter/BloomKFilterTest.java +++ b/extensions-core/druid-bloom-filter/src/test/java/org/apache/druid/query/filter/BloomKFilterTest.java @@ -52,28 +52,28 @@ public void testBloomKFilterBytes() throws IOException bf.add(val); BloomKFilter.add(buffer, val); - BloomKFilter rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + BloomKFilter rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.test(val)); Assert.assertFalse(rehydrated.test(val1)); Assert.assertFalse(rehydrated.test(val2)); Assert.assertFalse(rehydrated.test(val3)); BloomKFilter.add(buffer, val1); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.test(val)); Assert.assertTrue(rehydrated.test(val1)); Assert.assertFalse(rehydrated.test(val2)); Assert.assertFalse(rehydrated.test(val3)); BloomKFilter.add(buffer, val2); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.test(val)); Assert.assertTrue(rehydrated.test(val1)); Assert.assertTrue(rehydrated.test(val2)); Assert.assertFalse(rehydrated.test(val3)); BloomKFilter.add(buffer, val3); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.test(val)); Assert.assertTrue(rehydrated.test(val1)); @@ -86,7 +86,7 @@ public void testBloomKFilterBytes() throws IOException BloomKFilter.add(buffer, randVal); } // last value should be present - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); Assert.assertTrue(rehydrated.test(randVal)); // most likely this value should not exist randVal[0] = 0; @@ -114,28 +114,28 @@ public void testBloomKFilterByte() throws IOException byte val3 = Byte.MAX_VALUE; BloomKFilter.addLong(buffer, val); - BloomKFilter rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + BloomKFilter rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertFalse(rehydrated.testLong(val1)); Assert.assertFalse(rehydrated.testLong(val2)); Assert.assertFalse(rehydrated.testLong(val3)); BloomKFilter.addLong(buffer, val1); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertTrue(rehydrated.testLong(val1)); Assert.assertFalse(rehydrated.testLong(val2)); Assert.assertFalse(rehydrated.testLong(val3)); BloomKFilter.addLong(buffer, val2); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertTrue(rehydrated.testLong(val1)); Assert.assertTrue(rehydrated.testLong(val2)); Assert.assertFalse(rehydrated.testLong(val3)); BloomKFilter.addLong(buffer, val3); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertTrue(rehydrated.testLong(val1)); @@ -148,7 +148,7 @@ public void testBloomKFilterByte() throws IOException BloomKFilter.addLong(buffer, randVal); } - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); // last value should be present Assert.assertTrue(rehydrated.testLong(randVal)); @@ -173,28 +173,28 @@ public void testBloomKFilterInt() throws IOException int val3 = Integer.MAX_VALUE; BloomKFilter.addLong(buffer, val); - BloomKFilter rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + BloomKFilter rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertFalse(rehydrated.testLong(val1)); Assert.assertFalse(rehydrated.testLong(val2)); Assert.assertFalse(rehydrated.testLong(val3)); BloomKFilter.addLong(buffer, val1); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertTrue(rehydrated.testLong(val1)); Assert.assertFalse(rehydrated.testLong(val2)); Assert.assertFalse(rehydrated.testLong(val3)); BloomKFilter.addLong(buffer, val2); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertTrue(rehydrated.testLong(val1)); Assert.assertTrue(rehydrated.testLong(val2)); Assert.assertFalse(rehydrated.testLong(val3)); BloomKFilter.addLong(buffer, val3); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertTrue(rehydrated.testLong(val1)); @@ -206,7 +206,7 @@ public void testBloomKFilterInt() throws IOException randVal = rand.nextInt(); BloomKFilter.addLong(buffer, randVal); } - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); // last value should be present Assert.assertTrue(rehydrated.testLong(randVal)); // most likely this value should not exist @@ -230,28 +230,28 @@ public void testBloomKFilterLong() throws IOException long val3 = Long.MAX_VALUE; BloomKFilter.addLong(buffer, val); - BloomKFilter rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + BloomKFilter rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertFalse(rehydrated.testLong(val1)); Assert.assertFalse(rehydrated.testLong(val2)); Assert.assertFalse(rehydrated.testLong(val3)); BloomKFilter.addLong(buffer, val1); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertTrue(rehydrated.testLong(val1)); Assert.assertFalse(rehydrated.testLong(val2)); Assert.assertFalse(rehydrated.testLong(val3)); BloomKFilter.addLong(buffer, val2); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertTrue(rehydrated.testLong(val1)); Assert.assertTrue(rehydrated.testLong(val2)); Assert.assertFalse(rehydrated.testLong(val3)); BloomKFilter.addLong(buffer, val3); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testLong(val)); Assert.assertTrue(rehydrated.testLong(val1)); @@ -263,7 +263,7 @@ public void testBloomKFilterLong() throws IOException randVal = rand.nextInt(); BloomKFilter.addLong(buffer, randVal); } - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); // last value should be present Assert.assertTrue(rehydrated.testLong(randVal)); // most likely this value should not exist @@ -287,28 +287,28 @@ public void testBloomKFilterFloat() throws IOException float val3 = Float.POSITIVE_INFINITY; BloomKFilter.addFloat(buffer, val); - BloomKFilter rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + BloomKFilter rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testFloat(val)); Assert.assertFalse(rehydrated.testFloat(val1)); Assert.assertFalse(rehydrated.testFloat(val2)); Assert.assertFalse(rehydrated.testFloat(val3)); BloomKFilter.addFloat(buffer, val1); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testFloat(val)); Assert.assertTrue(rehydrated.testFloat(val1)); Assert.assertFalse(rehydrated.testFloat(val2)); Assert.assertFalse(rehydrated.testFloat(val3)); BloomKFilter.addFloat(buffer, val2); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testFloat(val)); Assert.assertTrue(rehydrated.testFloat(val1)); Assert.assertTrue(rehydrated.testFloat(val2)); Assert.assertFalse(rehydrated.testFloat(val3)); BloomKFilter.addFloat(buffer, val3); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testFloat(val)); Assert.assertTrue(rehydrated.testFloat(val1)); @@ -320,7 +320,7 @@ public void testBloomKFilterFloat() throws IOException randVal = rand.nextFloat(); BloomKFilter.addFloat(buffer, randVal); } - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); // last value should be present Assert.assertTrue(rehydrated.testFloat(randVal)); @@ -345,28 +345,28 @@ public void testBloomKFilterDouble() throws IOException double val3 = Double.POSITIVE_INFINITY; BloomKFilter.addDouble(buffer, val); - BloomKFilter rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + BloomKFilter rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testDouble(val)); Assert.assertFalse(rehydrated.testDouble(val1)); Assert.assertFalse(rehydrated.testDouble(val2)); Assert.assertFalse(rehydrated.testDouble(val3)); BloomKFilter.addDouble(buffer, val1); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testDouble(val)); Assert.assertTrue(rehydrated.testDouble(val1)); Assert.assertFalse(rehydrated.testDouble(val2)); Assert.assertFalse(rehydrated.testDouble(val3)); BloomKFilter.addDouble(buffer, val2); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testDouble(val)); Assert.assertTrue(rehydrated.testDouble(val1)); Assert.assertTrue(rehydrated.testDouble(val2)); Assert.assertFalse(rehydrated.testDouble(val3)); BloomKFilter.addDouble(buffer, val3); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testDouble(val)); Assert.assertTrue(rehydrated.testDouble(val1)); @@ -378,7 +378,7 @@ public void testBloomKFilterDouble() throws IOException randVal = rand.nextDouble(); BloomKFilter.addDouble(buffer, randVal); } - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); // last value should be present Assert.assertTrue(rehydrated.testDouble(randVal)); @@ -403,28 +403,28 @@ public void testBloomKFilterString() throws IOException String val3 = "cuckoo filter"; BloomKFilter.addString(buffer, val); - BloomKFilter rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + BloomKFilter rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testString(val)); Assert.assertFalse(rehydrated.testString(val1)); Assert.assertFalse(rehydrated.testString(val2)); Assert.assertFalse(rehydrated.testString(val3)); BloomKFilter.addString(buffer, val1); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testString(val)); Assert.assertTrue(rehydrated.testString(val1)); Assert.assertFalse(rehydrated.testString(val2)); Assert.assertFalse(rehydrated.testString(val3)); BloomKFilter.addString(buffer, val2); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testString(val)); Assert.assertTrue(rehydrated.testString(val1)); Assert.assertTrue(rehydrated.testString(val2)); Assert.assertFalse(rehydrated.testString(val3)); BloomKFilter.addString(buffer, val3); - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); buffer.position(0); Assert.assertTrue(rehydrated.testString(val)); Assert.assertTrue(rehydrated.testString(val1)); @@ -436,7 +436,7 @@ public void testBloomKFilterString() throws IOException randVal = rand.nextLong(); BloomKFilter.addString(buffer, Long.toString(randVal)); } - rehydrated = BloomKFilter.deserialize(new ByteBufferInputStream(buffer)); + rehydrated = deserializeBloomFilter(buffer); // last value should be present Assert.assertTrue(rehydrated.testString(Long.toString(randVal))); // most likely this value should not exist @@ -536,4 +536,11 @@ public void testCountBitBloomKFilterByteBuffersEmpty() throws Exception BloomKFilter.getNumSetBits(bufWithValues, 0) > BloomKFilter.getNumSetBits(bufWithNull, 0) ); } + + private static BloomKFilter deserializeBloomFilter(final ByteBuffer buffer) throws IOException + { + try (final ByteBufferInputStream inputStream = new ByteBufferInputStream(buffer)) { + return BloomKFilter.deserialize(inputStream); + } + } } diff --git a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentPullerTest.java b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentPullerTest.java index 289c60219afc..75c241f3c110 100644 --- a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentPullerTest.java +++ b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentPullerTest.java @@ -95,7 +95,8 @@ public void testGZUncompress() throws IOException, SegmentLoadingException final File tmpFile = temporaryFolder.newFile("gzTest.gz"); - try (OutputStream outputStream = new GZIPOutputStream(new FileOutputStream(tmpFile))) { + try (final FileOutputStream fileOutputStream = new FileOutputStream(tmpFile); + final OutputStream outputStream = new GZIPOutputStream(fileOutputStream)) { outputStream.write(value); } @@ -135,7 +136,8 @@ public void testGZUncompressOn4xxError() throws IOException final File tmpFile = temporaryFolder.newFile("gzTest.gz"); - try (OutputStream outputStream = new GZIPOutputStream(new FileOutputStream(tmpFile))) { + try (final FileOutputStream fileOutputStream = new FileOutputStream(tmpFile); + final OutputStream outputStream = new GZIPOutputStream(fileOutputStream)) { outputStream.write(value); } @@ -181,7 +183,8 @@ public void testGZUncompressOn5xxError() throws IOException, SegmentLoadingExcep final File tmpFile = temporaryFolder.newFile("gzTest.gz"); - try (OutputStream outputStream = new GZIPOutputStream(new FileOutputStream(tmpFile))) { + try (final FileOutputStream fileOutputStream = new FileOutputStream(tmpFile); + final OutputStream outputStream = new GZIPOutputStream(fileOutputStream)) { outputStream.write(value); } diff --git a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3TaskLogsTest.java b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3TaskLogsTest.java index 2516b62ba4c8..3da6bf53ce23 100644 --- a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3TaskLogsTest.java +++ b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3TaskLogsTest.java @@ -454,10 +454,11 @@ public void test_taskLog_fetch() throws IOException S3TaskLogs s3TaskLogs = getS3TaskLogs(); Optional inputStreamOptional = s3TaskLogs.streamTaskLog(KEY_1, 0); - String taskLogs = new BufferedReader( - new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) - .lines() - .collect(Collectors.joining("\n")); + final String taskLogs; + try (final BufferedReader reader = new BufferedReader( + new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8))) { + taskLogs = reader.lines().collect(Collectors.joining("\n")); + } Assert.assertEquals(LOG_CONTENTS, taskLogs); } @@ -483,10 +484,11 @@ public void test_taskLog_fetch_withRange() throws IOException S3TaskLogs s3TaskLogs = getS3TaskLogs(); Optional inputStreamOptional = s3TaskLogs.streamTaskLog(KEY_1, 1); - String taskLogs = new BufferedReader( - new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) - .lines() - .collect(Collectors.joining("\n")); + final String taskLogs; + try (final BufferedReader reader = new BufferedReader( + new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8))) { + taskLogs = reader.lines().collect(Collectors.joining("\n")); + } Assert.assertEquals(LOG_CONTENTS.substring(1), taskLogs); } @@ -512,10 +514,11 @@ public void test_taskLog_fetch_withNegativeRange() throws IOException S3TaskLogs s3TaskLogs = getS3TaskLogs(); Optional inputStreamOptional = s3TaskLogs.streamTaskLog(KEY_1, -1 * (LOG_CONTENTS.length() - 1)); - String taskLogs = new BufferedReader( - new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) - .lines() - .collect(Collectors.joining("\n")); + final String taskLogs; + try (final BufferedReader reader = new BufferedReader( + new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8))) { + taskLogs = reader.lines().collect(Collectors.joining("\n")); + } Assert.assertEquals(LOG_CONTENTS.substring(1), taskLogs); } @@ -541,10 +544,11 @@ public void test_report_fetch() throws IOException S3TaskLogs s3TaskLogs = getS3TaskLogs(); Optional inputStreamOptional = s3TaskLogs.streamTaskReports(KEY_1); - String report = new BufferedReader( - new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) - .lines() - .collect(Collectors.joining("\n")); + final String report; + try (final BufferedReader reader = new BufferedReader( + new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8))) { + report = reader.lines().collect(Collectors.joining("\n")); + } Assert.assertEquals(REPORT_CONTENTS, report); } @@ -569,10 +573,11 @@ public void test_status_fetch() throws IOException S3TaskLogs s3TaskLogs = getS3TaskLogs(); Optional inputStreamOptional = s3TaskLogs.streamTaskStatus(KEY_1); - String report = new BufferedReader( - new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8)) - .lines() - .collect(Collectors.joining("\n")); + final String report; + try (final BufferedReader reader = new BufferedReader( + new InputStreamReader(inputStreamOptional.get(), StandardCharsets.UTF_8))) { + report = reader.lines().collect(Collectors.joining("\n")); + } Assert.assertEquals(STATUS_CONTENTS, report); } diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HttpShuffleClientTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HttpShuffleClientTest.java index 34e2b33b30c9..913cd4ff37b9 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HttpShuffleClientTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HttpShuffleClientTest.java @@ -20,6 +20,7 @@ package org.apache.druid.indexing.common.task.batch.parallel; import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; import org.apache.druid.java.util.common.Intervals; import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.concurrent.Execs; @@ -185,19 +186,26 @@ private HttpShuffleClient mockClient(int numFailures) throws FileNotFoundExcepti if (numFailures == 0) { EasyMock.expect(httpClient.go(EasyMock.anyObject(), EasyMock.anyObject())) // should return different instances of input stream - .andReturn(Futures.immediateFuture(new FileInputStream(segmentFile))) - .andReturn(Futures.immediateFuture(new FileInputStream(segmentFile))); + .andReturn(openSegmentFile()) + .andReturn(openSegmentFile()); } else { EasyMock.expect(httpClient.go(EasyMock.anyObject(), EasyMock.anyObject())) .andReturn(Futures.immediateFailedFuture(new RuntimeException())).times(numFailures) // should return different instances of input stream - .andReturn(Futures.immediateFuture(new FileInputStream(segmentFile))) - .andReturn(Futures.immediateFuture(new FileInputStream(segmentFile))); + .andReturn(openSegmentFile()) + .andReturn(openSegmentFile()); } EasyMock.replay(httpClient); return new HttpShuffleClient(httpClient); } + private ListenableFuture openSegmentFile() throws FileNotFoundException + { + // Ownership passes through HttpShuffleClient to FileUtils.copyLarge, which closes the response stream. + // codeql[java/input-resource-leak] + return Futures.immediateFuture(new FileInputStream(segmentFile)); + } + private static class TestPartitionLocation extends GenericPartitionLocation { private TestPartitionLocation() diff --git a/indexing-service/src/test/java/org/apache/druid/metadata/SQLMetadataStorageActionHandlerTest.java b/indexing-service/src/test/java/org/apache/druid/metadata/SQLMetadataStorageActionHandlerTest.java index 604f682fb66a..84f958b56e68 100644 --- a/indexing-service/src/test/java/org/apache/druid/metadata/SQLMetadataStorageActionHandlerTest.java +++ b/indexing-service/src/test/java/org/apache/druid/metadata/SQLMetadataStorageActionHandlerTest.java @@ -51,6 +51,7 @@ import org.junit.Test; import java.sql.ResultSet; +import java.sql.Statement; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -454,9 +455,11 @@ private Integer getUnmigratedTaskCount() "SELECT COUNT(*) FROM %s WHERE type is NULL or group_id is NULL", entryTable ); - ResultSet resultSet = handle.getConnection().createStatement().executeQuery(sql); - resultSet.next(); - return resultSet.getInt(1); + try (final Statement statement = handle.getConnection().createStatement(); + final ResultSet resultSet = statement.executeQuery(sql)) { + resultSet.next(); + return resultSet.getInt(1); + } } ); } diff --git a/processing/src/main/java/org/apache/druid/java/util/common/io/smoosh/FileSmoosher.java b/processing/src/main/java/org/apache/druid/java/util/common/io/smoosh/FileSmoosher.java index f6166e65648c..4b84e4cd7c5b 100644 --- a/processing/src/main/java/org/apache/druid/java/util/common/io/smoosh/FileSmoosher.java +++ b/processing/src/main/java/org/apache/druid/java/util/common/io/smoosh/FileSmoosher.java @@ -447,7 +447,9 @@ public static class Outer implements SmooshedWriter this.outFile = outFile; this.maxLength = maxLength; - FileOutputStream outStream = closer.register(new FileOutputStream(outFile)); // lgtm [java/output-resource-leak] + // The closer owns this stream for the lifetime of the writer and releases it in close(). + // codeql[java/output-resource-leak] + final FileOutputStream outStream = closer.register(new FileOutputStream(outFile)); this.channel = closer.register(outStream.getChannel()); } diff --git a/processing/src/main/java/org/apache/druid/segment/file/SegmentFileBuilderV10.java b/processing/src/main/java/org/apache/druid/segment/file/SegmentFileBuilderV10.java index d5281ab1186c..230b705b4bb1 100644 --- a/processing/src/main/java/org/apache/druid/segment/file/SegmentFileBuilderV10.java +++ b/processing/src/main/java/org/apache/druid/segment/file/SegmentFileBuilderV10.java @@ -685,6 +685,8 @@ private static class ContainerWriter implements GatheringByteChannel this.file = file; this.bundle = bundle; this.maxSize = maxSize; + // The closer owns this stream for the lifetime of the writer and releases it in close(). + // codeql[java/output-resource-leak] final FileOutputStream outStream = closer.register(new FileOutputStream(file)); this.channel = closer.register(outStream.getChannel()); } diff --git a/processing/src/test/java/org/apache/druid/data/input/impl/RetryingInputStreamTest.java b/processing/src/test/java/org/apache/druid/data/input/impl/RetryingInputStreamTest.java index ac5cff961bb5..a4855b91d58d 100644 --- a/processing/src/test/java/org/apache/druid/data/input/impl/RetryingInputStreamTest.java +++ b/processing/src/test/java/org/apache/druid/data/input/impl/RetryingInputStreamTest.java @@ -128,15 +128,15 @@ public void testThrowsOnIOException() throws IOException public void testRetryOnCustomException() throws IOException { throwCustomExceptions = 1; - final RetryingInputStream retryingInputStream = new RetryingInputStream<>( + try (final RetryingInputStream retryingInputStream = new RetryingInputStream<>( testFile, objectOpenFunction, t -> t instanceof CustomException, MAX_RETRY, false - ); - - retryHelper(retryingInputStream); + )) { + retryHelper(retryingInputStream); + } Assertions.assertEquals(0, throwCustomExceptions); } @@ -168,15 +168,15 @@ public void testResumeAfterExceptions() throws IOException readBytesBeforeExceptions = 1000; throwCustomExceptions = 100; - final RetryingInputStream retryingInputStream = new RetryingInputStream<>( + try (final RetryingInputStream retryingInputStream = new RetryingInputStream<>( testFile, objectOpenFunction, t -> true, // always retry MAX_RETRY, false - ); - - retryHelper(retryingInputStream); + )) { + retryHelper(retryingInputStream); + } // Tried more than MAX_RETRY times because progress was being made. (MAX_RETRIES applies to each call individually.) Assertions.assertEquals(81, throwCustomExceptions); @@ -207,15 +207,15 @@ public void testIOExceptionNotRetriableRead() throws IOException { throwCustomExceptions = 1; throwIOExceptions = 1; - final RetryingInputStream retryingInputStream = new RetryingInputStream<>( + try (final RetryingInputStream retryingInputStream = new RetryingInputStream<>( testFile, objectOpenFunction, t -> t instanceof IOException || t instanceof CustomException, MAX_RETRY, false - ); - - retryHelper(retryingInputStream); + )) { + retryHelper(retryingInputStream); + } Assertions.assertEquals(0, throwCustomExceptions); Assertions.assertEquals(0, throwIOExceptions); @@ -242,13 +242,15 @@ public InputStream answer(InvocationOnMock invocation) throws Throwable } }).when(objectOpenFunction).open(any(), anyLong()); - new RetryingInputStream<>( + try (final RetryingInputStream ignored = new RetryingInputStream<>( testFile, objectOpenFunction, t -> t instanceof CustomException, MAX_RETRY, false - ); + )) { + // Construction itself exercises the retry behavior. + } verify(objectOpenFunction, times(3)).open(any(), anyLong()); Assertions.assertEquals(0, throwCustomExceptions); } diff --git a/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java b/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java index 937c51458d90..69244cc298a6 100644 --- a/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java +++ b/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java @@ -75,10 +75,12 @@ public class CompressionUtilsTest static { final StringBuilder builder = new StringBuilder(); - try (InputStream stream = CompressionUtilsTest.class.getClassLoader().getResourceAsStream("white-rabbit.txt")) { - final Iterator it = new Scanner( - new InputStreamReader(stream, StandardCharsets.UTF_8) - ).useDelimiter(Pattern.quote(System.lineSeparator())); + try ( + final InputStream stream = CompressionUtilsTest.class.getClassLoader().getResourceAsStream("white-rabbit.txt"); + final InputStreamReader reader = new InputStreamReader(stream, StandardCharsets.UTF_8); + final Scanner scanner = new Scanner(reader).useDelimiter(Pattern.quote(System.lineSeparator())) + ) { + final Iterator it = scanner; while (it.hasNext()) { builder.append(it.next()); } @@ -193,7 +195,10 @@ public void testGoodGZCompressUncompressToFile() throws Exception Assert.assertFalse(gzFile.exists()); CompressionUtils.gzip(testFile, gzFile); Assert.assertTrue(gzFile.exists()); - try (final InputStream inputStream = new GZIPInputStream(new FileInputStream(gzFile))) { + try ( + final InputStream fileInputStream = new FileInputStream(gzFile); + final InputStream inputStream = new GZIPInputStream(fileInputStream) + ) { assertGoodDataStream(inputStream); } testFile.delete(); @@ -210,10 +215,15 @@ public void testGoodZipStream() throws IOException { final File tmpDir = temporaryFolder.newFolder("testGoodZipStream"); final File zipFile = new File(tmpDir, "compressionUtilTest.zip"); - CompressionUtils.zip(testDir, new FileOutputStream(zipFile)); + try (final OutputStream outputStream = new FileOutputStream(zipFile)) { + CompressionUtils.zip(testDir, outputStream); + } final File newDir = new File(tmpDir, "newDir"); newDir.mkdir(); - final FileUtils.FileCopyResult result = CompressionUtils.unzip(new FileInputStream(zipFile), newDir); + final FileUtils.FileCopyResult result; + try (final InputStream inputStream = new FileInputStream(zipFile)) { + result = CompressionUtils.unzip(inputStream, newDir); + } verifyUnzip(newDir, result, ImmutableMap.of(testFile.getName(), StringUtils.toUtf8(CONTENT))); } @@ -232,7 +242,9 @@ private Map writeZipWithManyFiles(final File zipFile) throws IOE } } - CompressionUtils.zip(srcDir, new FileOutputStream(zipFile)); + try (final OutputStream outputStream = new FileOutputStream(zipFile)) { + CompressionUtils.zip(srcDir, outputStream); + } return expectedFiles; } @@ -308,7 +320,10 @@ public void testGoodGzipByteSource() throws IOException Assert.assertFalse(gzFile.exists()); CompressionUtils.gzip(Files.asByteSource(testFile), Files.asByteSink(gzFile), Predicates.alwaysTrue()); Assert.assertTrue(gzFile.exists()); - try (final InputStream inputStream = CompressionUtils.decompress(new FileInputStream(gzFile), gzFile.getName())) { + try ( + final InputStream fileInputStream = new FileInputStream(gzFile); + final InputStream inputStream = CompressionUtils.decompress(fileInputStream, gzFile.getName()) + ) { assertGoodDataStream(inputStream); } if (!testFile.delete()) { @@ -328,10 +343,17 @@ public void testDecompressBzip2() throws IOException final File tmpDir = temporaryFolder.newFolder("testDecompressBzip2"); final File bzFile = new File(tmpDir, testFile.getName() + ".bz2"); Assert.assertFalse(bzFile.exists()); - try (final OutputStream out = new BZip2CompressorOutputStream(new FileOutputStream(bzFile))) { - ByteStreams.copy(new FileInputStream(testFile), out); + try ( + final OutputStream fileOutputStream = new FileOutputStream(bzFile); + final OutputStream out = new BZip2CompressorOutputStream(fileOutputStream); + final InputStream in = new FileInputStream(testFile) + ) { + ByteStreams.copy(in, out); } - try (final InputStream inputStream = CompressionUtils.decompress(new FileInputStream(bzFile), bzFile.getName())) { + try ( + final InputStream fileInputStream = new FileInputStream(bzFile); + final InputStream inputStream = CompressionUtils.decompress(fileInputStream, bzFile.getName()) + ) { assertGoodDataStream(inputStream); } } @@ -342,10 +364,17 @@ public void testDecompressXz() throws IOException final File tmpDir = temporaryFolder.newFolder("testDecompressXz"); final File xzFile = new File(tmpDir, testFile.getName() + ".xz"); Assert.assertFalse(xzFile.exists()); - try (final OutputStream out = new XZCompressorOutputStream(new FileOutputStream(xzFile))) { - ByteStreams.copy(new FileInputStream(testFile), out); + try ( + final OutputStream fileOutputStream = new FileOutputStream(xzFile); + final OutputStream out = new XZCompressorOutputStream(fileOutputStream); + final InputStream in = new FileInputStream(testFile) + ) { + ByteStreams.copy(in, out); } - try (final InputStream inputStream = CompressionUtils.decompress(new FileInputStream(xzFile), xzFile.getName())) { + try ( + final InputStream fileInputStream = new FileInputStream(xzFile); + final InputStream inputStream = CompressionUtils.decompress(fileInputStream, xzFile.getName()) + ) { assertGoodDataStream(inputStream); } } @@ -356,10 +385,17 @@ public void testDecompressSnappy() throws IOException final File tmpDir = temporaryFolder.newFolder("testDecompressSnappy"); final File snappyFile = new File(tmpDir, testFile.getName() + ".sz"); Assert.assertFalse(snappyFile.exists()); - try (final OutputStream out = new FramedSnappyCompressorOutputStream(new FileOutputStream(snappyFile))) { - ByteStreams.copy(new FileInputStream(testFile), out); + try ( + final OutputStream fileOutputStream = new FileOutputStream(snappyFile); + final OutputStream out = new FramedSnappyCompressorOutputStream(fileOutputStream); + final InputStream in = new FileInputStream(testFile) + ) { + ByteStreams.copy(in, out); } - try (final InputStream inputStream = CompressionUtils.decompress(new FileInputStream(snappyFile), snappyFile.getName())) { + try ( + final InputStream fileInputStream = new FileInputStream(snappyFile); + final InputStream inputStream = CompressionUtils.decompress(fileInputStream, snappyFile.getName()) + ) { assertGoodDataStream(inputStream); } } @@ -370,10 +406,17 @@ public void testDecompressZstd() throws IOException final File tmpDir = temporaryFolder.newFolder("testDecompressZstd"); final File zstdFile = new File(tmpDir, testFile.getName() + ".zst"); Assert.assertFalse(zstdFile.exists()); - try (final OutputStream out = new ZstdCompressorOutputStream(new FileOutputStream(zstdFile))) { - ByteStreams.copy(new FileInputStream(testFile), out); + try ( + final OutputStream fileOutputStream = new FileOutputStream(zstdFile); + final OutputStream out = new ZstdCompressorOutputStream(fileOutputStream); + final InputStream in = new FileInputStream(testFile) + ) { + ByteStreams.copy(in, out); } - try (final InputStream inputStream = CompressionUtils.decompress(new FileInputStream(zstdFile), zstdFile.getName())) { + try ( + final InputStream fileInputStream = new FileInputStream(zstdFile); + final InputStream inputStream = CompressionUtils.decompress(fileInputStream, zstdFile.getName()) + ) { assertGoodDataStream(inputStream); } } @@ -384,12 +427,19 @@ public void testDecompressZip() throws IOException final File tmpDir = temporaryFolder.newFolder("testDecompressZip"); final File zipFile = new File(tmpDir, testFile.getName() + ".zip"); Assert.assertFalse(zipFile.exists()); - try (final ZipOutputStream out = new ZipOutputStream(new FileOutputStream(zipFile))) { + try ( + final OutputStream fileOutputStream = new FileOutputStream(zipFile); + final ZipOutputStream out = new ZipOutputStream(fileOutputStream); + final InputStream in = new FileInputStream(testFile) + ) { out.putNextEntry(new ZipEntry("cool.file")); - ByteStreams.copy(new FileInputStream(testFile), out); + ByteStreams.copy(in, out); out.closeEntry(); } - try (final InputStream inputStream = CompressionUtils.decompress(new FileInputStream(zipFile), zipFile.getName())) { + try ( + final InputStream fileInputStream = new FileInputStream(zipFile); + final InputStream inputStream = CompressionUtils.decompress(fileInputStream, zipFile.getName()) + ) { assertGoodDataStream(inputStream); } } @@ -401,7 +451,10 @@ public void testDecompressZipWithManyFiles() throws IOException final File zipFile = new File(tmpDir, testFile.getName() + ".zip"); writeZipWithManyFiles(zipFile); - try (final InputStream inputStream = CompressionUtils.decompress(new FileInputStream(zipFile), zipFile.getName())) { + try ( + final InputStream fileInputStream = new FileInputStream(zipFile); + final InputStream inputStream = CompressionUtils.decompress(fileInputStream, zipFile.getName()) + ) { // Should read the first file, which contains a single null byte. Assert.assertArrayEquals(new byte[]{0}, ByteStreams.toByteArray(inputStream)); } @@ -453,16 +506,26 @@ public void testGoodGZStream() throws IOException final File tmpDir = temporaryFolder.newFolder("testGoodGZStream"); final File gzFile = new File(tmpDir, testFile.getName() + ".gz"); Assert.assertFalse(gzFile.exists()); - CompressionUtils.gzip(new FileInputStream(testFile), new FileOutputStream(gzFile)); + try ( + final InputStream inputStream = new FileInputStream(testFile); + final OutputStream outputStream = new FileOutputStream(gzFile) + ) { + CompressionUtils.gzip(inputStream, outputStream); + } Assert.assertTrue(gzFile.exists()); - try (final InputStream inputStream = new GZIPInputStream(new FileInputStream(gzFile))) { + try ( + final InputStream fileInputStream = new FileInputStream(gzFile); + final InputStream inputStream = new GZIPInputStream(fileInputStream) + ) { assertGoodDataStream(inputStream); } if (!testFile.delete()) { throw new IOE("Unable to delete file [%s]", testFile.getAbsolutePath()); } Assert.assertFalse(testFile.exists()); - CompressionUtils.gunzip(new FileInputStream(gzFile), testFile); + try (final InputStream inputStream = new FileInputStream(gzFile)) { + CompressionUtils.gunzip(inputStream, testFile); + } Assert.assertTrue(testFile.exists()); try (final InputStream inputStream = new FileInputStream(testFile)) { assertGoodDataStream(inputStream); @@ -504,8 +567,8 @@ public void testEvilZipInputStream() throws IOException java.nio.file.Files.deleteIfExists(evilZip.toPath()); CompressionUtilsTest.makeEvilZip(evilZip); - try { - CompressionUtils.unzip(new FileInputStream(evilZip), tmpDir); + try (final InputStream inputStream = new FileInputStream(evilZip)) { + CompressionUtils.unzip(inputStream, tmpDir); } catch (ISE ise) { Assert.assertTrue(ise.getMessage().contains("does not start with outDir")); @@ -741,7 +804,10 @@ public void flush() throws IOException }, Predicates.alwaysTrue() ); Assert.assertTrue(gzFile.exists()); - try (final InputStream inputStream = CompressionUtils.decompress(new FileInputStream(gzFile), "file.gz")) { + try ( + final InputStream fileInputStream = new FileInputStream(gzFile); + final InputStream inputStream = CompressionUtils.decompress(fileInputStream, "file.gz") + ) { assertGoodDataStream(inputStream); } if (!testFile.delete()) { @@ -763,8 +829,9 @@ public void testStreamErrorGzip() throws Exception final File gzFile = new File(tmpDir, testFile.getName() + ".gz"); Assert.assertFalse(gzFile.exists()); final AtomicLong flushes = new AtomicLong(0L); - CompressionUtils.gzip( - new FileInputStream(testFile), new FileOutputStream(gzFile) + try ( + final InputStream inputStream = new FileInputStream(testFile); + final OutputStream outputStream = new FileOutputStream(gzFile) { @Override public void flush() throws IOException @@ -776,7 +843,9 @@ public void flush() throws IOException } } } - ); + ) { + CompressionUtils.gzip(inputStream, outputStream); + } } @Test(expected = IOException.class) @@ -787,7 +856,10 @@ public void testStreamErrorGunzip() throws Exception Assert.assertFalse(gzFile.exists()); CompressionUtils.gzip(Files.asByteSource(testFile), Files.asByteSink(gzFile), Predicates.alwaysTrue()); Assert.assertTrue(gzFile.exists()); - try (final InputStream inputStream = CompressionUtils.decompress(new FileInputStream(gzFile), "file.gz")) { + try ( + final InputStream fileInputStream = new FileInputStream(gzFile); + final InputStream inputStream = CompressionUtils.decompress(fileInputStream, "file.gz") + ) { assertGoodDataStream(inputStream); } if (testFile.exists() && !testFile.delete()) { @@ -795,22 +867,24 @@ public void testStreamErrorGunzip() throws Exception } Assert.assertFalse(testFile.exists()); final AtomicLong flushes = new AtomicLong(0L); - CompressionUtils.gunzip( - new FileInputStream(gzFile), new FilterOutputStream( - new FileOutputStream(testFile) - { - @Override - public void flush() throws IOException - { - if (flushes.getAndIncrement() > 0) { - super.flush(); - } else { - throw new IOException("Test exception"); - } - } + try ( + final InputStream inputStream = new FileInputStream(gzFile); + final OutputStream fileOutputStream = new FileOutputStream(testFile) + { + @Override + public void flush() throws IOException + { + if (flushes.getAndIncrement() > 0) { + super.flush(); + } else { + throw new IOException("Test exception"); } - ) - ); + } + }; + final OutputStream outputStream = new FilterOutputStream(fileOutputStream) + ) { + CompressionUtils.gunzip(inputStream, outputStream); + } } private void verifyUnzip( diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/AggregationTestHelper.java b/processing/src/test/java/org/apache/druid/query/aggregation/AggregationTestHelper.java index b8a31fe584b8..9269a04b1a2b 100644 --- a/processing/src/test/java/org/apache/druid/query/aggregation/AggregationTestHelper.java +++ b/processing/src/test/java/org/apache/druid/query/aggregation/AggregationTestHelper.java @@ -366,17 +366,19 @@ public void createIndex( int maxRowCount ) throws Exception { - createIndex( - new FileInputStream(inputDataFile), - inputSchema, - inputFormat, - aggregators, - outDir, - minTimestamp, - gran, - maxRowCount, - true - ); + try (final InputStream inputStream = new FileInputStream(inputDataFile)) { + createIndex( + inputStream, + inputSchema, + inputFormat, + aggregators, + outDir, + minTimestamp, + gran, + maxRowCount, + true + ); + } } public void createIndex( @@ -391,17 +393,19 @@ public void createIndex( boolean rollup ) throws Exception { - createIndex( - new FileInputStream(inputDataFile), - inputSchema, - inputFormat, - aggregators, - outDir, - minTimestamp, - gran, - maxRowCount, - rollup - ); + try (final InputStream inputStream = new FileInputStream(inputDataFile)) { + createIndex( + inputStream, + inputSchema, + inputFormat, + aggregators, + outDir, + minTimestamp, + gran, + maxRowCount, + rollup + ); + } } public void createIndex( @@ -704,4 +708,3 @@ public void close() throws IOException resourceCloser.close(); } } - diff --git a/server/src/test/java/org/apache/druid/metadata/input/SqlEntityTest.java b/server/src/test/java/org/apache/druid/metadata/input/SqlEntityTest.java index 4053ec9ecf22..f86abf123266 100644 --- a/server/src/test/java/org/apache/druid/metadata/input/SqlEntityTest.java +++ b/server/src/test/java/org/apache/druid/metadata/input/SqlEntityTest.java @@ -66,15 +66,17 @@ public void testExecuteQuery() throws IOException SqlTestUtils testUtils = new SqlTestUtils(derbyConnector); final InputRow expectedRow = testUtils.createTableWithRows(TABLE_NAME_1, 1).get(0); File tmpFile = File.createTempFile("testQueryResults", ""); - InputEntity.CleanableFile queryResult = SqlEntity.openCleanableFile( + final String actualJson; + try (final InputEntity.CleanableFile queryResult = SqlEntity.openCleanableFile( VALID_SQL, testUtils.getDerbyInputSourceConnector(), mapper, true, tmpFile ); - InputStream queryInputStream = new FileInputStream(queryResult.file()); - String actualJson = IOUtils.toString(queryInputStream, StandardCharsets.UTF_8); + final InputStream queryInputStream = new FileInputStream(queryResult.file())) { + actualJson = IOUtils.toString(queryInputStream, StandardCharsets.UTF_8); + } String expectedJson = mapper.writeValueAsString( Collections.singletonList(((MapBasedInputRow) expectedRow).getEvent()) ); diff --git a/server/src/test/java/org/apache/druid/rpc/RequestBuilderTest.java b/server/src/test/java/org/apache/druid/rpc/RequestBuilderTest.java index ca95284ac565..03ca63d6198e 100644 --- a/server/src/test/java/org/apache/druid/rpc/RequestBuilderTest.java +++ b/server/src/test/java/org/apache/druid/rpc/RequestBuilderTest.java @@ -129,10 +129,12 @@ public void test_build_postTlsWithContent() throws Exception Assert.assertTrue(request.hasContent()); // Read and verify content. - Assert.assertEquals( - json, - StringUtils.fromUtf8(ByteStreams.toByteArray(new ChannelBufferInputStream(request.getContent()))) - ); + try (final ChannelBufferInputStream inputStream = new ChannelBufferInputStream(request.getContent())) { + Assert.assertEquals( + json, + StringUtils.fromUtf8(ByteStreams.toByteArray(inputStream)) + ); + } } @Test @@ -151,10 +153,12 @@ public void test_build_postTlsWithJsonContent() throws Exception Assert.assertTrue(request.hasContent()); // Read and verify content. - Assert.assertEquals( - "{\"foo\":3}", - StringUtils.fromUtf8(ByteStreams.toByteArray(new ChannelBufferInputStream(request.getContent()))) - ); + try (final ChannelBufferInputStream inputStream = new ChannelBufferInputStream(request.getContent())) { + Assert.assertEquals( + "{\"foo\":3}", + StringUtils.fromUtf8(ByteStreams.toByteArray(inputStream)) + ); + } } @Test diff --git a/server/src/test/java/org/apache/druid/server/QueryResourceTest.java b/server/src/test/java/org/apache/druid/server/QueryResourceTest.java index 7922588cf777..ad8a1986236c 100644 --- a/server/src/test/java/org/apache/druid/server/QueryResourceTest.java +++ b/server/src/test/java/org/apache/druid/server/QueryResourceTest.java @@ -1039,11 +1039,16 @@ public void testIncompleteQuery() throws IOException @Test public void testResourceLimitExceeded() throws IOException { - Response response = queryResource.doPost( - new ExceptionalInputStream(() -> new ResourceLimitExceededException("You require too much of something")), - null /*pretty*/, - testServletRequest - ); + final Response response; + try (final ExceptionalInputStream inputStream = new ExceptionalInputStream( + () -> new ResourceLimitExceededException("You require too much of something") + )) { + response = queryResource.doPost( + inputStream, + null /*pretty*/, + testServletRequest + ); + } Assert.assertNotNull(response); Assert.assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus()); QueryException e = jsonMapper.readValue((byte[]) response.getEntity(), QueryException.class); @@ -1055,11 +1060,16 @@ public void testResourceLimitExceeded() throws IOException public void testUnsupportedQueryThrowsException() throws IOException { String errorMessage = "This will be support in Druid 9999"; - Response response = queryResource.doPost( - new ExceptionalInputStream(() -> new QueryUnsupportedException(errorMessage)), - null /*pretty*/, - testServletRequest - ); + final Response response; + try (final ExceptionalInputStream inputStream = new ExceptionalInputStream( + () -> new QueryUnsupportedException(errorMessage) + )) { + response = queryResource.doPost( + inputStream, + null /*pretty*/, + testServletRequest + ); + } Assert.assertNotNull(response); Assert.assertEquals(QueryUnsupportedException.STATUS_CODE, response.getStatus()); QueryException ex = jsonMapper.readValue((byte[]) response.getEntity(), QueryException.class); diff --git a/server/src/test/java/org/apache/druid/server/initialization/JettyTest.java b/server/src/test/java/org/apache/druid/server/initialization/JettyTest.java index d2f3c5078bc5..67b56318093d 100644 --- a/server/src/test/java/org/apache/druid/server/initialization/JettyTest.java +++ b/server/src/test/java/org/apache/druid/server/initialization/JettyTest.java @@ -290,31 +290,39 @@ public void testGzipResponseCompression() throws Exception final HttpURLConnection get = (HttpURLConnection) url.openConnection(); get.setRequestProperty("Accept-Encoding", "gzip"); Assert.assertEquals("gzip", get.getContentEncoding()); - Assert.assertEquals( - DEFAULT_RESPONSE_CONTENT, - IOUtils.toString(new GZIPInputStream(get.getInputStream()), StandardCharsets.UTF_8) - ); + try (final InputStream inputStream = new GZIPInputStream(get.getInputStream())) { + Assert.assertEquals( + DEFAULT_RESPONSE_CONTENT, + IOUtils.toString(inputStream, StandardCharsets.UTF_8) + ); + } final HttpURLConnection post = (HttpURLConnection) url.openConnection(); post.setRequestProperty("Accept-Encoding", "gzip"); post.setRequestMethod("POST"); Assert.assertEquals("gzip", post.getContentEncoding()); - Assert.assertEquals( - DEFAULT_RESPONSE_CONTENT, - IOUtils.toString(new GZIPInputStream(post.getInputStream()), StandardCharsets.UTF_8) - ); + try (final InputStream inputStream = new GZIPInputStream(post.getInputStream())) { + Assert.assertEquals( + DEFAULT_RESPONSE_CONTENT, + IOUtils.toString(inputStream, StandardCharsets.UTF_8) + ); + } final HttpURLConnection getNoGzip = (HttpURLConnection) url.openConnection(); Assert.assertNotEquals("gzip", getNoGzip.getContentEncoding()); - Assert.assertEquals(DEFAULT_RESPONSE_CONTENT, IOUtils.toString(getNoGzip.getInputStream(), StandardCharsets.UTF_8)); + try (final InputStream inputStream = getNoGzip.getInputStream()) { + Assert.assertEquals(DEFAULT_RESPONSE_CONTENT, IOUtils.toString(inputStream, StandardCharsets.UTF_8)); + } final HttpURLConnection postNoGzip = (HttpURLConnection) url.openConnection(); postNoGzip.setRequestMethod("POST"); Assert.assertNotEquals("gzip", postNoGzip.getContentEncoding()); - Assert.assertEquals( - DEFAULT_RESPONSE_CONTENT, - IOUtils.toString(postNoGzip.getInputStream(), StandardCharsets.UTF_8) - ); + try (final InputStream inputStream = postNoGzip.getInputStream()) { + Assert.assertEquals( + DEFAULT_RESPONSE_CONTENT, + IOUtils.toString(inputStream, StandardCharsets.UTF_8) + ); + } } // Tests that threads are not stuck when partial chunk is not finalized diff --git a/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java b/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java index 20c46f3a1082..54dffb5f0bbf 100644 --- a/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java +++ b/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java @@ -963,7 +963,9 @@ public void testConcurrentQueries() @Test public void testTooManyStatements() throws SQLException { + // Leave these statements open until tearDown closes client so the test reaches the configured limit. for (int i = 0; i < STATEMENT_LIMIT; i++) { + // codeql[java/database-resource-leak] client.createStatement(); } @@ -978,7 +980,9 @@ public void testTooManyStatements() throws SQLException public void testNotTooManyStatementsWhenYouCloseThem() throws SQLException { for (int i = 0; i < STATEMENT_LIMIT * 2; i++) { - client.createStatement().close(); + try (final Statement ignored = client.createStatement()) { + // Closing each statement is the behavior under test. + } } } @@ -1037,7 +1041,7 @@ public void tesErrorsDoNotCloseStatements() throws SQLException public void testNotTooManyStatementsWhenClosed() { for (int i = 0; i < 50; i++) { - try (Statement statement = client.createStatement()) { + try (final Statement statement = client.createStatement()) { statement.executeQuery("SELECT SUM(nonexistent) FROM druid.foo"); Assert.fail(); } @@ -1051,11 +1055,13 @@ public void testNotTooManyStatementsWhenClosed() public void testAutoReconnectOnNoSuchConnection() throws SQLException { for (int i = 0; i < 50; i++) { - final ResultSet resultSet = client.createStatement().executeQuery("SELECT COUNT(*) AS cnt FROM druid.foo"); - Assert.assertEquals( - ImmutableList.of(ImmutableMap.of("cnt", 6L)), - getRows(resultSet) - ); + try (Statement statement = client.createStatement()) { + final ResultSet resultSet = statement.executeQuery("SELECT COUNT(*) AS cnt FROM druid.foo"); + Assert.assertEquals( + ImmutableList.of(ImmutableMap.of("cnt", 6L)), + getRows(resultSet) + ); + } server.druidMeta.closeAllConnections(); } } @@ -1063,9 +1069,14 @@ public void testAutoReconnectOnNoSuchConnection() throws SQLException @Test public void testTooManyConnections() throws SQLException { + // Keep one statement open on each connection until tearDown so all connection slots remain occupied. + // codeql[java/database-resource-leak] client.createStatement(); + // codeql[java/database-resource-leak] clientLosAngeles.createStatement(); + // codeql[java/database-resource-leak] superuserClient.createStatement(); + // codeql[java/database-resource-leak] clientNoTrailingSlash.createStatement(); AvaticaClientRuntimeException ex = Assert.assertThrows( @@ -1090,11 +1101,15 @@ public void testConnectionsCloseStatements() throws SQLException for (int i = 0; i < CONNECTION_LIMIT * 2; i++) { try (Connection connection = server.getUserConnection()) { // Note: NOT in a try-catch block. Let the connection close the statement + // codeql[java/database-resource-leak] final Statement statement = connection.createStatement(); // Again, NOT in a try-catch block: let the statement close the // result set. - final ResultSet resultSet = statement.executeQuery("SELECT COUNT(*) AS cnt FROM druid.foo"); + // codeql[java/database-resource-leak] + final ResultSet resultSet = statement.executeQuery( + "SELECT COUNT(*) AS cnt FROM druid.foo" + ); Assert.assertTrue(resultSet.next()); } } @@ -1158,27 +1173,24 @@ public Frame fetch( }; ServerWrapper server = new ServerWrapper(smallFrameDruidMeta); - Connection smallFrameClient = server.getUserConnection(); - - final ResultSet resultSet = smallFrameClient.createStatement().executeQuery( - "SELECT dim1 FROM druid.foo" - ); - List> rows = getRows(resultSet); - Assert.assertEquals(2, frames.size()); - Assert.assertEquals( - ImmutableList.of( - ImmutableMap.of("dim1", ""), - ImmutableMap.of("dim1", "10.1"), - ImmutableMap.of("dim1", "2"), - ImmutableMap.of("dim1", "1"), - ImmutableMap.of("dim1", "def"), - ImmutableMap.of("dim1", "abc") - ), - rows - ); + try (final Connection smallFrameClient = server.getUserConnection(); + final Statement statement = smallFrameClient.createStatement(); + final ResultSet resultSet = statement.executeQuery("SELECT dim1 FROM druid.foo")) { + final List> rows = getRows(resultSet); + Assert.assertEquals(2, frames.size()); + Assert.assertEquals( + ImmutableList.of( + ImmutableMap.of("dim1", ""), + ImmutableMap.of("dim1", "10.1"), + ImmutableMap.of("dim1", "2"), + ImmutableMap.of("dim1", "1"), + ImmutableMap.of("dim1", "def"), + ImmutableMap.of("dim1", "abc") + ), + rows + ); + } - resultSet.close(); - smallFrameClient.close(); exec.shutdown(); server.close(); } @@ -1220,30 +1232,29 @@ public Frame fetch( }; ServerWrapper server = new ServerWrapper(smallFrameDruidMeta); - Connection smallFrameClient = server.getUserConnection(); - - // use a prepared statement because Avatica currently ignores fetchSize on the initial fetch of a Statement - PreparedStatement statement = smallFrameClient.prepareStatement("SELECT dim1 FROM druid.foo"); - // set a fetch size below the minimum configured threshold - statement.setFetchSize(2); - final ResultSet resultSet = statement.executeQuery(); - List> rows = getRows(resultSet); - // expect minimum threshold to be used, which should be enough to do this all in first fetch - Assert.assertEquals(0, frames.size()); - Assert.assertEquals( - ImmutableList.of( - ImmutableMap.of("dim1", ""), - ImmutableMap.of("dim1", "10.1"), - ImmutableMap.of("dim1", "2"), - ImmutableMap.of("dim1", "1"), - ImmutableMap.of("dim1", "def"), - ImmutableMap.of("dim1", "abc") - ), - rows - ); + try (final Connection smallFrameClient = server.getUserConnection(); + final PreparedStatement statement = smallFrameClient.prepareStatement("SELECT dim1 FROM druid.foo")) { + // use a prepared statement because Avatica currently ignores fetchSize on the initial fetch of a Statement + // set a fetch size below the minimum configured threshold + statement.setFetchSize(2); + try (final ResultSet resultSet = statement.executeQuery()) { + final List> rows = getRows(resultSet); + // expect minimum threshold to be used, which should be enough to do this all in first fetch + Assert.assertEquals(0, frames.size()); + Assert.assertEquals( + ImmutableList.of( + ImmutableMap.of("dim1", ""), + ImmutableMap.of("dim1", "10.1"), + ImmutableMap.of("dim1", "2"), + ImmutableMap.of("dim1", "1"), + ImmutableMap.of("dim1", "def"), + ImmutableMap.of("dim1", "abc") + ), + rows + ); + } + } - resultSet.close(); - smallFrameClient.close(); exec.shutdown(); server.close(); } @@ -1834,8 +1845,8 @@ public Frame fetch( try (Connection conn = server.getUserConnection()) { // Test with plain JDBC - try (ResultSet resultSet = conn.createStatement().executeQuery( - "SELECT dim1 FROM druid.foo")) { + try (final Statement statement = conn.createStatement(); + final ResultSet resultSet = statement.executeQuery("SELECT dim1 FROM druid.foo")) { List> rows = getRows(resultSet); Assert.assertEquals(6, rows.size()); Assert.assertEquals(6, frames.size()); // 3 empty frames and then 3 frames of 2 rows each From a4619616f0bf226279678e8e56ef09325630626d Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Fri, 31 Jul 2026 06:45:07 +0800 Subject: [PATCH 2/4] style: make Avatica statement final --- .../org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java b/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java index 54dffb5f0bbf..3742001a34d3 100644 --- a/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java +++ b/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java @@ -1055,7 +1055,7 @@ public void testNotTooManyStatementsWhenClosed() public void testAutoReconnectOnNoSuchConnection() throws SQLException { for (int i = 0; i < 50; i++) { - try (Statement statement = client.createStatement()) { + try (final Statement statement = client.createStatement()) { final ResultSet resultSet = statement.executeQuery("SELECT COUNT(*) AS cnt FROM druid.foo"); Assert.assertEquals( ImmutableList.of(ImmutableMap.of("cnt", 6L)), From 39c3a1d864cd14333940f437ce70057e7a5ce81d Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Fri, 31 Jul 2026 07:05:21 +0800 Subject: [PATCH 3/4] test: ensure cleanup after Avatica failures --- .../util/common/CompressionUtilsTest.java | 25 +++---- .../sql/avatica/DruidAvaticaHandlerTest.java | 66 ++++++++++--------- 2 files changed, 47 insertions(+), 44 deletions(-) diff --git a/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java b/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java index 69244cc298a6..c922b4c435ec 100644 --- a/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java +++ b/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java @@ -869,19 +869,20 @@ public void testStreamErrorGunzip() throws Exception final AtomicLong flushes = new AtomicLong(0L); try ( final InputStream inputStream = new FileInputStream(gzFile); - final OutputStream fileOutputStream = new FileOutputStream(testFile) - { - @Override - public void flush() throws IOException - { - if (flushes.getAndIncrement() > 0) { - super.flush(); - } else { - throw new IOException("Test exception"); + final OutputStream outputStream = new FilterOutputStream( + new FileOutputStream(testFile) + { + @Override + public void flush() throws IOException + { + if (flushes.getAndIncrement() > 0) { + super.flush(); + } else { + throw new IOException("Test exception"); + } + } } - } - }; - final OutputStream outputStream = new FilterOutputStream(fileOutputStream) + ) ) { CompressionUtils.gunzip(inputStream, outputStream); } diff --git a/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java b/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java index 3742001a34d3..416f04624c13 100644 --- a/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java +++ b/sql/src/test/java/org/apache/druid/sql/avatica/DruidAvaticaHandlerTest.java @@ -206,7 +206,7 @@ private DruidSchemaCatalog makeRootSchema() ); } - private class ServerWrapper + private class ServerWrapper implements AutoCloseable { final DruidMeta druidMeta; final Server server; @@ -248,6 +248,7 @@ public Connection getUserConnection() throws SQLException // return DriverManager.getConnection(url); //} + @Override public void close() throws Exception { druidMeta.closeAllConnections(); @@ -1172,8 +1173,8 @@ public Frame fetch( } }; - ServerWrapper server = new ServerWrapper(smallFrameDruidMeta); - try (final Connection smallFrameClient = server.getUserConnection(); + try (final ServerWrapper server = new ServerWrapper(smallFrameDruidMeta); + final Connection smallFrameClient = server.getUserConnection(); final Statement statement = smallFrameClient.createStatement(); final ResultSet resultSet = statement.executeQuery("SELECT dim1 FROM druid.foo")) { final List> rows = getRows(resultSet); @@ -1190,9 +1191,9 @@ public Frame fetch( rows ); } - - exec.shutdown(); - server.close(); + finally { + exec.shutdown(); + } } @Test @@ -1231,8 +1232,8 @@ public Frame fetch( } }; - ServerWrapper server = new ServerWrapper(smallFrameDruidMeta); - try (final Connection smallFrameClient = server.getUserConnection(); + try (final ServerWrapper server = new ServerWrapper(smallFrameDruidMeta); + final Connection smallFrameClient = server.getUserConnection(); final PreparedStatement statement = smallFrameClient.prepareStatement("SELECT dim1 FROM druid.foo")) { // use a prepared statement because Avatica currently ignores fetchSize on the initial fetch of a Statement // set a fetch size below the minimum configured threshold @@ -1254,9 +1255,9 @@ public Frame fetch( ); } } - - exec.shutdown(); - server.close(); + finally { + exec.shutdown(); + } } @Test @@ -1841,29 +1842,30 @@ public Frame fetch( } }; - ServerWrapper server = new ServerWrapper(druidMeta); - try (Connection conn = server.getUserConnection()) { - - // Test with plain JDBC - try (final Statement statement = conn.createStatement(); - final ResultSet resultSet = statement.executeQuery("SELECT dim1 FROM druid.foo")) { - List> rows = getRows(resultSet); - Assert.assertEquals(6, rows.size()); - Assert.assertEquals(6, frames.size()); // 3 empty frames and then 3 frames of 2 rows each - - Assert.assertFalse(frames.get(0).rows.iterator().hasNext()); - Assert.assertFalse(frames.get(1).rows.iterator().hasNext()); - Assert.assertFalse(frames.get(2).rows.iterator().hasNext()); - Assert.assertTrue(frames.get(3).rows.iterator().hasNext()); - Assert.assertTrue(frames.get(4).rows.iterator().hasNext()); - Assert.assertTrue(frames.get(5).rows.iterator().hasNext()); + try (final ServerWrapper server = new ServerWrapper(druidMeta)) { + try (final Connection conn = server.getUserConnection()) { + + // Test with plain JDBC + try (final Statement statement = conn.createStatement(); + final ResultSet resultSet = statement.executeQuery("SELECT dim1 FROM druid.foo")) { + final List> rows = getRows(resultSet); + Assert.assertEquals(6, rows.size()); + Assert.assertEquals(6, frames.size()); // 3 empty frames and then 3 frames of 2 rows each + + Assert.assertFalse(frames.get(0).rows.iterator().hasNext()); + Assert.assertFalse(frames.get(1).rows.iterator().hasNext()); + Assert.assertFalse(frames.get(2).rows.iterator().hasNext()); + Assert.assertTrue(frames.get(3).rows.iterator().hasNext()); + Assert.assertTrue(frames.get(4).rows.iterator().hasNext()); + Assert.assertTrue(frames.get(5).rows.iterator().hasNext()); + } } - } - - testWithJDBI(server.url); - exec.shutdown(); - server.close(); + testWithJDBI(server.url); + } + finally { + exec.shutdown(); + } } @Test From e04124afaec9f15db7b4fd2768bd6b4aa739d2ea Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Fri, 31 Jul 2026 07:10:44 +0800 Subject: [PATCH 4/4] test: tighten resource fixture types --- .../task/batch/parallel/HttpShuffleClientTest.java | 10 ++++++---- .../druid/java/util/common/CompressionUtilsTest.java | 6 +++++- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HttpShuffleClientTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HttpShuffleClientTest.java index 913cd4ff37b9..b0e1ca12aaaa 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HttpShuffleClientTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HttpShuffleClientTest.java @@ -25,6 +25,7 @@ import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.concurrent.Execs; import org.apache.druid.java.util.http.client.HttpClient; +import org.apache.druid.java.util.http.client.response.InputStreamResponseHandler; import org.apache.druid.utils.CompressionUtils; import org.easymock.EasyMock; import org.joda.time.Interval; @@ -39,6 +40,7 @@ import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.IOException; +import java.io.InputStream; import java.io.Writer; import java.nio.charset.StandardCharsets; import java.nio.file.Files; @@ -184,12 +186,12 @@ private HttpShuffleClient mockClient(int numFailures) throws FileNotFoundExcepti { HttpClient httpClient = EasyMock.strictMock(HttpClient.class); if (numFailures == 0) { - EasyMock.expect(httpClient.go(EasyMock.anyObject(), EasyMock.anyObject())) + EasyMock.expect(httpClient.go(EasyMock.anyObject(), EasyMock.anyObject())) // should return different instances of input stream .andReturn(openSegmentFile()) .andReturn(openSegmentFile()); } else { - EasyMock.expect(httpClient.go(EasyMock.anyObject(), EasyMock.anyObject())) + EasyMock.expect(httpClient.go(EasyMock.anyObject(), EasyMock.anyObject())) .andReturn(Futures.immediateFailedFuture(new RuntimeException())).times(numFailures) // should return different instances of input stream .andReturn(openSegmentFile()) @@ -199,11 +201,11 @@ private HttpShuffleClient mockClient(int numFailures) throws FileNotFoundExcepti return new HttpShuffleClient(httpClient); } - private ListenableFuture openSegmentFile() throws FileNotFoundException + private ListenableFuture openSegmentFile() throws FileNotFoundException { // Ownership passes through HttpShuffleClient to FileUtils.copyLarge, which closes the response stream. // codeql[java/input-resource-leak] - return Futures.immediateFuture(new FileInputStream(segmentFile)); + return Futures.immediateFuture(new FileInputStream(segmentFile)); } private static class TestPartitionLocation extends GenericPartitionLocation diff --git a/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java b/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java index c922b4c435ec..50a65c568322 100644 --- a/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java +++ b/processing/src/test/java/org/apache/druid/java/util/common/CompressionUtilsTest.java @@ -57,6 +57,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Scanner; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -76,7 +77,10 @@ public class CompressionUtilsTest static { final StringBuilder builder = new StringBuilder(); try ( - final InputStream stream = CompressionUtilsTest.class.getClassLoader().getResourceAsStream("white-rabbit.txt"); + final InputStream stream = Objects.requireNonNull( + CompressionUtilsTest.class.getClassLoader().getResourceAsStream("white-rabbit.txt"), + "Missing test resource: white-rabbit.txt" + ); final InputStreamReader reader = new InputStreamReader(stream, StandardCharsets.UTF_8); final Scanner scanner = new Scanner(reader).useDelimiter(Pattern.quote(System.lineSeparator())) ) {