This is an automated email from the ASF dual-hosted git repository. ppkarwasz pushed a commit to branch feat/logging-parent-migration2 in repository https://gitbox.apache.org/repos/asf/logging-flume.git
commit 78ca669e21c8fc07df390c29923b3d2e55a6d548 Author: Piotr P. Karwasz <[email protected]> AuthorDate: Fri Jul 24 14:13:54 2026 +0200 Fix or suppress Error Prone warnings The `GuardedBy` suppressions are intentional: the channel queue locks must not be held on blocking operations and the unguarded reads are benign. The `jvm.config` flags let Error Prone access javac internals on JDK 16+. Assisted-By: Claude Fable 5 <[email protected]> --- .mvn/jvm.config | 11 +++++++++++ .../apache/flume/channel/SpillableMemoryChannel.java | 9 +++++++++ .../java/org/apache/flume/channel/MemoryChannel.java | 12 ++++++++++++ .../java/org/apache/flume/event/EventHelper.java | 2 +- .../org/apache/flume/channel/TestMemoryChannel.java | 3 +++ .../flume/channel/TestMemoryChannelConcurrency.java | 2 +- .../org/apache/flume/source/http/TestHTTPSource.java | 20 +++++++------------- 7 files changed, 44 insertions(+), 15 deletions(-) diff --git a/.mvn/jvm.config b/.mvn/jvm.config new file mode 100644 index 00000000..f3b9f7db --- /dev/null +++ b/.mvn/jvm.config @@ -0,0 +1,11 @@ +--add-exports jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED +--add-exports jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED +--add-exports jdk.compiler/com.sun.tools.javac.main=ALL-UNNAMED +--add-exports jdk.compiler/com.sun.tools.javac.model=ALL-UNNAMED +--add-exports jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED +--add-exports jdk.compiler/com.sun.tools.javac.processing=ALL-UNNAMED +--add-exports jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED +--add-exports jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED +--add-opens jdk.compiler/com.sun.tools.javac.code=ALL-UNNAMED +--add-opens jdk.compiler/com.sun.tools.javac.comp=ALL-UNNAMED +-Dfile.encoding=UTF-8 diff --git a/flume-ng-channels/flume-spillable-memory-channel/src/main/java/org/apache/flume/channel/SpillableMemoryChannel.java b/flume-ng-channels/flume-spillable-memory-channel/src/main/java/org/apache/flume/channel/SpillableMemoryChannel.java index 36bdf192..25c8ada3 100644 --- a/flume-ng-channels/flume-spillable-memory-channel/src/main/java/org/apache/flume/channel/SpillableMemoryChannel.java +++ b/flume-ng-channels/flume-spillable-memory-channel/src/main/java/org/apache/flume/channel/SpillableMemoryChannel.java @@ -535,6 +535,9 @@ public class SpillableMemoryChannel extends FileChannel { channelCounter.addToEventPutSuccessCount(putList.size()); } + // `queueLock` must not be held on blocking operations, + // and an approximate channel size is enough for the counter. + @SuppressWarnings("GuardedBy") @Override protected void doRollback() { logger.debug("Rollback() of " + (takeCalled ? " Take Tx" : (putCalled ? " Put Tx" : "Empty Tx"))); @@ -729,6 +732,8 @@ public class SpillableMemoryChannel extends FileChannel { super.configure(context); } + // the unguarded read of `memQueue` only detects whether the queue needs resizing + @SuppressWarnings("GuardedBy") private void resizePrimaryQueue(int newMemoryCapacity) throws InterruptedException { if (memQueue != null && memoryCapacity == newMemoryCapacity) { return; @@ -765,6 +770,8 @@ public class SpillableMemoryChannel extends FileChannel { } } + // an approximate channel size is enough for the counter + @SuppressWarnings("GuardedBy") @Override public synchronized void start() { super.start(); @@ -779,6 +786,8 @@ public class SpillableMemoryChannel extends FileChannel { channelCounter.setChannelSize(totalCount); } + // an approximate channel size is enough for the counter + @SuppressWarnings("GuardedBy") @Override public synchronized void stop() { if (getLifecycleState() == LifecycleState.STOP) { diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java b/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java index ac3c8cbe..b0c6c3c6 100644 --- a/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java +++ b/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java @@ -110,6 +110,9 @@ public class MemoryChannel extends BasicChannelSemantics implements TransactionC return event; } + // `queueLock` must not be held on blocking operations, + // and an approximate channel size is sufficient for the counter. + @SuppressWarnings("GuardedBy") @Override protected void doCommit() throws InterruptedException { int remainingChange = takeList.size() - putList.size(); @@ -158,6 +161,9 @@ public class MemoryChannel extends BasicChannelSemantics implements TransactionC channelCounter.setChannelSize(queue.size()); } + // `queueLock` must not be held on blocking operations + // and an approximate channel size is sufficient for the counter. + @SuppressWarnings("GuardedBy") @Override protected void doRollback() { int takes = takeList.size(); @@ -218,6 +224,8 @@ public class MemoryChannel extends BasicChannelSemantics implements TransactionC * <li>byteCapacityBufferPercentage = type int that defines the percent of buffer between byteCapacity and the estimated event size. * <li>keep-alive = type int that defines the number of second to wait for a queue permit */ + // the unguarded null check on `queue` only detects whether the channel was already configured + @SuppressWarnings("GuardedBy") @Override public void configure(Context context) { Integer capacity = null; @@ -347,6 +355,8 @@ public class MemoryChannel extends BasicChannelSemantics implements TransactionC } } + // an approximate channel size is sufficient for the counter + @SuppressWarnings("GuardedBy") @Override public synchronized void start() { channelCounter.start(); @@ -355,6 +365,8 @@ public class MemoryChannel extends BasicChannelSemantics implements TransactionC super.start(); } + // an approximate channel size is sufficient for the counter + @SuppressWarnings("GuardedBy") @Override public synchronized void stop() { channelCounter.setChannelSize(queue.size()); diff --git a/flume-ng-core/src/main/java/org/apache/flume/event/EventHelper.java b/flume-ng-core/src/main/java/org/apache/flume/event/EventHelper.java index 58a010b1..733377f2 100644 --- a/flume-ng-core/src/main/java/org/apache/flume/event/EventHelper.java +++ b/flume-ng-core/src/main/java/org/apache/flume/event/EventHelper.java @@ -61,7 +61,7 @@ public class EventHelper { } String result = buffer.toString(); if (result.endsWith(EOL) && buffer.length() > EOL.length()) { - buffer.delete(buffer.length() - EOL.length(), buffer.length()).toString(); + buffer.delete(buffer.length() - EOL.length(), buffer.length()); } } return "{ headers:" + event.getHeaders() + " body:" + buffer + " }"; diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java index 99877351..166f8dac 100644 --- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java +++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java @@ -34,6 +34,7 @@ import org.apache.flume.event.SimpleEvent; import org.apache.flume.exception.ChannelException; import org.junit.Assert; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; public class TestMemoryChannel { @@ -298,6 +299,8 @@ public class TestMemoryChannel { Assert.assertEquals(8, channel.getBytesRemainingValue()); } + @Test + @Ignore public void testByteCapacityBufferEmptyingAfterTakeCommit() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java index 693a62ad..8ee407a7 100644 --- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java +++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java @@ -224,6 +224,7 @@ public class TestMemoryChannelConcurrency { final CountDownLatch startGate = new CountDownLatch(1); final CountDownLatch endGate = new CountDownLatch(threadCount); + final Object takeMapLock = new Object(); // start a sink and source for each for (int i = 0; i < threadCount / 2; i++) { Thread t = new Thread() { @@ -268,7 +269,6 @@ public class TestMemoryChannelConcurrency { }; // start source t.start(); - final Integer takeMapLock = 0; t = new Thread() { @Override public void run() { diff --git a/flume-ng-sources/flume-http-source/src/test/java/org/apache/flume/source/http/TestHTTPSource.java b/flume-ng-sources/flume-http-source/src/test/java/org/apache/flume/source/http/TestHTTPSource.java index 42b63f6a..8aa5ea2e 100644 --- a/flume-ng-sources/flume-http-source/src/test/java/org/apache/flume/source/http/TestHTTPSource.java +++ b/flume-ng-sources/flume-http-source/src/test/java/org/apache/flume/source/http/TestHTTPSource.java @@ -385,8 +385,8 @@ public class TestHTTPSource { .size() == 0); - int newPort = findFreePort(); - Context configuredSourceContext = getDefaultNonSecureContext(newPort); + final int firstPort = findFreePort(); + Context configuredSourceContext = getDefaultNonSecureContext(firstPort); configuredSourceContext.put("HttpConfiguration.sendServerVersion", "false"); configuredSourceContext.put("HttpConfiguration.sendXPoweredBy", "true"); configuredSourceContext.put("ServerConnector.acceptQueueSize", "22"); @@ -398,7 +398,7 @@ public class TestHTTPSource { newChannel.start(); newSource.start(); - HttpPost newPostRequest = new HttpPost("http://0.0.0.0:"; + newPort); + HttpPost newPostRequest = new HttpPost("http://0.0.0.0:"; + firstPort); resp = httpClient.execute(newPostRequest); Assert.assertTrue(resp.getHeaders("X-Powered-By").length > 0); @@ -414,8 +414,8 @@ public class TestHTTPSource { newChannel.stop(); // Configure SslContextFactory with junk protocols (expect failure) - newPort = findFreePort(); - configuredSourceContext = getDefaultSecureContext(newPort); + final int secondPort = findFreePort(); + configuredSourceContext = getDefaultSecureContext(secondPort); configuredSourceContext.put("SslContextFactory.IncludeProtocols", "abc def"); newSource = new HTTPSource(); @@ -426,14 +426,8 @@ public class TestHTTPSource { newChannel.start(); newSource.start(); - newPostRequest = new HttpPost("http://0.0.0.0:"; + newPort); - try { - doTestHttps(null, newPort, httpsChannel); - // We are testing that this fails because we've deliberately configured the wrong protocols - Assert.assertTrue(false); - } catch (AssertionError ex) { - // no-op - } + newPostRequest = new HttpPost("http://0.0.0.0:"; + secondPort); + Assert.assertThrows(AssertionError.class, () -> doTestHttps(null, secondPort, httpsChannel)); newSource.stop(); newChannel.stop(); }
