diff --git a/build.gradle b/build.gradle index a61cd9f1d7432..d00745c9f5390 100644 --- a/build.gradle +++ b/build.gradle @@ -424,6 +424,8 @@ subprojects { maxHeapSize = defaultMaxHeapSize jvmArgs = defaultJvmArgs + systemProperty "junit.jupiter.extensions.autodetection.enabled", "true" + testLogging { events = userTestLoggingEvents ?: testLoggingEvents showStandardStreams = userShowStandardStreams ?: testShowStandardStreams @@ -456,6 +458,7 @@ subprojects { maxHeapSize = "2560m" jvmArgs = defaultJvmArgs + systemProperty "junit.jupiter.extensions.autodetection.enabled", "true" testLogging { events = userTestLoggingEvents ?: testLoggingEvents @@ -472,7 +475,7 @@ subprojects { useJUnitPlatform { includeTags "integration" includeTags "org.apache.kafka.test.IntegrationTest" - // Both engines are needed to run JUnit 4 tests alongside JUnit 5 tests. + // Both engines are needed to run JUnit 4 tests alongside JUnit 5 tests. // junit-vintage (JUnit 4) can be removed once the JUnit 4 migration is complete. includeEngines "junit-vintage", "junit-jupiter" } @@ -505,6 +508,8 @@ subprojects { maxHeapSize = defaultMaxHeapSize jvmArgs = defaultJvmArgs + systemProperty "junit.jupiter.extensions.autodetection.enabled", "true" + testLogging { events = userTestLoggingEvents ?: testLoggingEvents showStandardStreams = userShowStandardStreams ?: testShowStandardStreams @@ -520,7 +525,7 @@ subprojects { useJUnitPlatform { excludeTags "integration" excludeTags "org.apache.kafka.test.IntegrationTest" - // Both engines are needed to run JUnit 4 tests alongside JUnit 5 tests. + // Both engines are needed to run JUnit 4 tests alongside JUnit 5 tests. // junit-vintage (JUnit 4) can be removed once the JUnit 4 migration is complete. includeEngines "junit-vintage", "junit-jupiter" } diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Exit.java b/clients/src/main/java/org/apache/kafka/common/utils/Exit.java index 20a7b2035ca16..827ef86c37917 100644 --- a/clients/src/main/java/org/apache/kafka/common/utils/Exit.java +++ b/clients/src/main/java/org/apache/kafka/common/utils/Exit.java @@ -54,8 +54,25 @@ public void addShutdownHook(String name, Runnable runnable) { } }; - private volatile static Procedure exitProcedure = DEFAULT_EXIT_PROCEDURE; - private volatile static Procedure haltProcedure = DEFAULT_HALT_PROCEDURE; + // The procedures to use if no custom procedure has been installed via setExitProcedure/setHaltProcedure + private volatile static Procedure fallbackExitProcedure = DEFAULT_EXIT_PROCEDURE; + private volatile static Procedure fallbackHaltProcedure = DEFAULT_HALT_PROCEDURE; + + // Use InheritableThreadLocal so that all threads use the same custom procedure(s), but + // once new test cases are started, leaked threads from prior tests don't have those custom procedures + // overwritten by the new cases, and don't accidentally invoke the custom procedures for the new cases + private static final InheritableThreadLocal EXIT_PROCEDURE = new InheritableThreadLocal() { + @Override + protected ProcedureWithFallback initialValue() { + return ProcedureWithFallback.forExit(); + } + }; + private static final InheritableThreadLocal HALT_PROCEDURE = new InheritableThreadLocal() { + @Override + protected ProcedureWithFallback initialValue() { + return ProcedureWithFallback.forHalt(); + } + }; private volatile static ShutdownHookAdder shutdownHookAdder = DEFAULT_SHUTDOWN_HOOK_ADDER; public static void exit(int statusCode) { @@ -63,7 +80,7 @@ public static void exit(int statusCode) { } public static void exit(int statusCode, String message) { - exitProcedure.execute(statusCode, message); + EXIT_PROCEDURE.get().procedure().execute(statusCode, message); } public static void halt(int statusCode) { @@ -71,19 +88,29 @@ public static void halt(int statusCode) { } public static void halt(int statusCode, String message) { - haltProcedure.execute(statusCode, message); + HALT_PROCEDURE.get().procedure().execute(statusCode, message); } public static void addShutdownHook(String name, Runnable runnable) { shutdownHookAdder.addShutdownHook(name, runnable); } + public static void setFallbackExitProcedure(Procedure procedure) { + fallbackExitProcedure = procedure; + EXIT_PROCEDURE.get().setFallback(procedure); + } + + public static void setFallbackHaltProcedure(Procedure procedure) { + fallbackHaltProcedure = procedure; + HALT_PROCEDURE.get().setFallback(procedure); + } + public static void setExitProcedure(Procedure procedure) { - exitProcedure = procedure; + EXIT_PROCEDURE.get().setCustom(procedure); } public static void setHaltProcedure(Procedure procedure) { - haltProcedure = procedure; + HALT_PROCEDURE.get().setCustom(procedure); } public static void setShutdownHookAdder(ShutdownHookAdder shutdownHookAdder) { @@ -91,14 +118,55 @@ public static void setShutdownHookAdder(ShutdownHookAdder shutdownHookAdder) { } public static void resetExitProcedure() { - exitProcedure = DEFAULT_EXIT_PROCEDURE; + EXIT_PROCEDURE.get().useFallback(); + EXIT_PROCEDURE.set(ProcedureWithFallback.forExit()); } public static void resetHaltProcedure() { - haltProcedure = DEFAULT_HALT_PROCEDURE; + HALT_PROCEDURE.get().useFallback(); + HALT_PROCEDURE.set(ProcedureWithFallback.forHalt()); } public static void resetShutdownHookAdder() { shutdownHookAdder = DEFAULT_SHUTDOWN_HOOK_ADDER; } + + private static class ProcedureWithFallback { + private volatile Procedure fallback; + private volatile Procedure custom; + private volatile boolean useCustom; + + public static ProcedureWithFallback forExit() { + return new ProcedureWithFallback(fallbackExitProcedure); + } + + public static ProcedureWithFallback forHalt() { + return new ProcedureWithFallback(fallbackHaltProcedure); + } + + private ProcedureWithFallback(Procedure fallbackProcedure) { + this.fallback = fallbackProcedure; + this.custom = null; + this.useCustom = true; + } + + public void setCustom(Procedure procedure) { + this.custom = procedure; + } + + public void setFallback(Procedure procedure) { + this.fallback = procedure; + } + + // Disable any custom procedures and force the fallback to be used; should be invoked + // once a test case has completed, in order to prevent leaked threads from that test case + // from calling the custom procedure + public void useFallback() { + useCustom = false; + } + + public Procedure procedure() { + return custom != null && useCustom ? custom : fallback; + } + } } diff --git a/core/src/test/java/kafka/test/junit/SystemExitExtension.java b/core/src/test/java/kafka/test/junit/SystemExitExtension.java new file mode 100644 index 0000000000000..85199799cc739 --- /dev/null +++ b/core/src/test/java/kafka/test/junit/SystemExitExtension.java @@ -0,0 +1,219 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.test.junit; + +import org.apache.kafka.common.utils.Exit; +import org.junit.jupiter.api.extension.AfterEachCallback; +import org.junit.jupiter.api.extension.BeforeEachCallback; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Method; +import java.util.Collection; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +public class SystemExitExtension implements BeforeEachCallback, AfterEachCallback { + + private static final Logger log = LoggerFactory.getLogger(SystemExitExtension.class); + private static final Map ACTIVE_TEST_CASES = new ConcurrentHashMap<>(); + private static final Map FINISHED_TEST_CASES = new ConcurrentHashMap<>(); + private static final InheritableThreadLocal CURRENT_TEST_INSTANCE = new InheritableThreadLocal<>(); + + public SystemExitExtension() { + log.debug("Instantiating system exit extension to guard against unintentional calls to terminate the JVM"); + } + + @Override + public void beforeEach(ExtensionContext context) { + addActiveTestCase(context); + } + + @Override + public void afterEach(ExtensionContext context) { + finishActiveTestCase(context); + checkForLateExits(); + } + + /** + * For testing the extension; should not be used by any other tests + */ + static void allowExitFromCurrentTest() { + Object testInstance = CURRENT_TEST_INSTANCE.get(); + assertNotNull(testInstance); + TestCase testCase = ACTIVE_TEST_CASES.get(testInstance); + assertNotNull(testCase); + testCase.allowExit(true); + } + + /** + * For testing the extension; should not be used by any other tests + */ + static void assertExitCalledFromCurrentTest(int expectedStatus) { + Object testInstance = CURRENT_TEST_INSTANCE.get(); + assertNotNull(testInstance); + TestCase testCase = ACTIVE_TEST_CASES.remove(testInstance); + assertExitFromTest(testCase, expectedStatus); + } + + /** + * For testing the extension; should not be used by any other tests + */ + static void assertExitFromPriorTest(Object testInstance, int expectedStatus) { + TestCase testCase = FINISHED_TEST_CASES.remove(testInstance); + assertExitFromTest(testCase, expectedStatus); + } + + private static void assertExitFromTest(TestCase testCase, int expectedStatus) { + assertNotNull(testCase); + Integer actualStatus = testCase.exit(); + assertEquals(Integer.valueOf(expectedStatus), actualStatus); + } + + private static void addActiveTestCase(ExtensionContext context) { + Object testInstance = context.getRequiredTestInstance(); + CURRENT_TEST_INSTANCE.set(testInstance); + Method testMethod = context.getRequiredTestMethod(); + String testCaseDescription = testMethod.getDeclaringClass() + "::" + testMethod.getName(); + ACTIVE_TEST_CASES.put(testInstance, new TestCase(testCaseDescription)); + Exit.setFallbackExitProcedure(exitHandler(testInstance)); + Exit.setFallbackHaltProcedure(exitHandler(testInstance)); + } + + private static void finishActiveTestCase(ExtensionContext context) { + Exit.resetExitProcedure(); + Exit.resetHaltProcedure(); + // Any leaked threads from the test case will continue to see the value that they were instantiated with, + // but this thread should be reset since it may be reused for newer tests + CURRENT_TEST_INSTANCE.remove(); + + Object testInstance = context.getRequiredTestInstance(); + synchronized (SystemExitExtension.class) { + TestCase testCase = ACTIVE_TEST_CASES.remove(testInstance); + if (testCase == null) { + // Should never happen, but just in case... + return; + } + if (testCase.exit() != null && !testCase.allowExit()) { + throw new AssertionError("Exit/halt was invoked with status " + testCase.exit() + " during this test. " + + "This test should either be modified to install a custom exit/halt procedure in the Exit class, or " + + "if no calls to Exit::exit or Exit::halt are expected, fixed to prevent these calls from taking place."); + } else { + // Continue tracking tests that have completed without exiting as they may leak threads + // that try to exit later on + FINISHED_TEST_CASES.put(testInstance, testCase); + } + } + } + + private static void checkForLateExits() { + Collection lateExits; + synchronized (SystemExitExtension.class) { + lateExits = FINISHED_TEST_CASES.values().stream() + .filter(tc -> tc.exit() != null) + .filter(tc -> !tc.allowExit()) + .collect(Collectors.toSet()); + FINISHED_TEST_CASES.values().removeAll(lateExits); + } + if (!lateExits.isEmpty()) { + throw new AssertionError( + "Exit/halt was invoked by threads spawned for testing after those tests had completed; " + + "this test will fail in order to surface these illegal calls. The calls occurred in the following tests:\n" + + lateExits.stream().map(Object::toString).collect(Collectors.joining("\n")) + + "Since the attempts to terminate the JVM originated from potentially-leaked threads, it may not be sufficient to " + + "use the Exit wrapper class, since its behavior must be reset at the end of each test, at which point other threads " + + "spawned during testing may attempt to use it. The offending test may have to be modified or disabled." + ); + } + } + + private static class TestCase { + + private volatile boolean allowExit; + private final String description; + private final AtomicReference exit; + + public TestCase(String description) { + this.allowExit = false; + this.description = Objects.requireNonNull(description); + this.exit = new AtomicReference<>(); + } + + public void recordExit(int status) { + exit.compareAndSet(null, status); + } + + public Integer exit() { + return exit.get(); + } + + // For testing only { + public boolean allowExit() { + return allowExit; + } + + // For testing only + public void allowExit(boolean allowExit) { + this.allowExit = allowExit; + } + + @Override + public String toString() { + String result = description; + if (exit() != null) { + result += " (exited with status " + exit() + ")"; + } + return result; + } + } + + private static org.apache.kafka.common.utils.Exit.Procedure exitHandler(Object testInstance) { + return (statusCode, message) -> { + TestCase testCase; + synchronized (SystemExitExtension.class) { + testCase = ACTIVE_TEST_CASES.get(testInstance); + if (testCase == null) { + testCase = FINISHED_TEST_CASES.get(testInstance); + } + if (testCase == null) { + // In this case, it's possible that a single test has caused multiple attempts to terminate the JVM + // to take place, and we've already removed it from our collection of both active and finished test + // cases since we only need to report at most one attempt per test + return; + } + + testCase.recordExit(statusCode); + } + + String errorMessage = "Exit or halt was invoked during test with status " + statusCode; + if (message != null) { + errorMessage += " and message '" + message + "'"; + } + + throw new AssertionError(errorMessage); + }; + } + +} diff --git a/core/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/core/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension new file mode 100644 index 0000000000000..c69f280d7bfc1 --- /dev/null +++ b/core/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension @@ -0,0 +1,15 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +kafka.test.junit.SystemExitExtension diff --git a/core/src/test/scala/kafka/test/junit/SystemExitExtensionTest.scala b/core/src/test/scala/kafka/test/junit/SystemExitExtensionTest.scala new file mode 100644 index 0000000000000..3f9994064c415 --- /dev/null +++ b/core/src/test/scala/kafka/test/junit/SystemExitExtensionTest.scala @@ -0,0 +1,124 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package kafka.test.junit + +import kafka.utils.Exit +import org.junit.jupiter.api.Assertions.{assertInstanceOf, assertNotNull, assertThrows, assertTrue} +import org.junit.jupiter.api.{MethodOrderer, Order, Test, TestMethodOrder} + +import java.util.concurrent.{CountDownLatch, TimeUnit} +import java.util.concurrent.atomic.AtomicReference + +object SystemExitExtensionTest { + private val lateExitStatus = 923 + private val lateExitOrigin = new AtomicReference[Object] + + @volatile private var lateExitReady: CountDownLatch = _ + @volatile private var lateExitInvoked: CountDownLatch = _ +} + +@TestMethodOrder(classOf[MethodOrderer.OrderAnnotation]) +class SystemExitExtensionTest { + + @Test + @Order(1) + def testSystemExitInvokedOnMainTestThread(): Unit = { + SystemExitExtension.allowExitFromCurrentTest() + + val exitStatus = 618 + // The caller should know that the attempt to terminate the JVM has failed + assertThrows(classOf[AssertionError], () => Exit.exit(exitStatus)) + // The extension should track that exit has been called in this test as well + SystemExitExtension.assertExitCalledFromCurrentTest(exitStatus) + } + + @Test + @Order(2) + def testSystemExitInvokedOnSeparateTestThread(): Unit = { + SystemExitExtension.allowExitFromCurrentTest() + + val exitStatus = 815 + val exitInvoked = new CountDownLatch(1) + val exitException = new AtomicReference[Throwable]() + new Thread(() => { + try { + Exit.exit(exitStatus) + } catch { + case t: Throwable => exitException.set(t) + } finally { + exitInvoked.countDown() + } + }).start() + + awaitLatch( + exitInvoked, + "Thread created during this test did not complete execution in time" + ) + // Wait for our separate thread to complete + exitInvoked.await(10, TimeUnit.SECONDS) + + // The caller should know that the attempt to terminate the JVM has failed + assertNotNull(exitException.get) + assertInstanceOf(classOf[AssertionError], exitException.get) + + // The extension should track that exit has been called in this test as well + SystemExitExtension.assertExitCalledFromCurrentTest(exitStatus) + } + + @Test + @Order(3) + // This test doesn't actually make any assertions; all it does is "leak" a thread that + // then invokes System::exit later on while a different test is running, so that that + // other test can ensure that the call was properly handled and attributed to this one + def spawnLateExitingTest(): Unit = { + SystemExitExtension.allowExitFromCurrentTest() + + SystemExitExtensionTest.lateExitOrigin.set(this) + SystemExitExtensionTest.lateExitReady = new CountDownLatch(1) + SystemExitExtensionTest.lateExitInvoked = new CountDownLatch(1) + + new Thread(() => { + // Await this latch indefinitely + SystemExitExtensionTest.lateExitReady.await() + try { + Exit.exit(SystemExitExtensionTest.lateExitStatus) + } finally { + SystemExitExtensionTest.lateExitInvoked.countDown() + } + }).start() + } + + @Test + @Order(4) + // This test ensures that the "leaked" thread from the above test is properly + // handled and attributed to that test + def testSystemExitInvokedOnLeakedTestThread(): Unit = { + assertNotNull(SystemExitExtensionTest.lateExitOrigin.get()) + SystemExitExtensionTest.lateExitReady.countDown() + awaitLatch( + SystemExitExtensionTest.lateExitInvoked, + "Leaked thread from prior test did not complete execution in time" + ) + + SystemExitExtension.assertExitFromPriorTest(SystemExitExtensionTest.lateExitOrigin.get(), SystemExitExtensionTest.lateExitStatus) + } + + private def awaitLatch(latch: CountDownLatch, assertionMessage: String): Unit = { + assertTrue(latch.await(10, TimeUnit.SECONDS), assertionMessage) + } + +}