[GitHub] [kafka] rhauch commented on a change in pull request #8259: KAFKA-7421: Ensure Connect's PluginClassLoader is truly parallel capable and resolve deadlock occurrences
rhauch commented on a change in pull request #8259: URL: https://github.com/apache/kafka/pull/8259#discussion_r589553615 ## File path: connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/SynchronizationTest.java ## @@ -0,0 +1,468 @@ +/* + * 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 org.apache.kafka.connect.runtime.isolation; + +import static org.junit.Assert.fail; + +import java.lang.management.LockInfo; +import java.lang.management.ManagementFactory; +import java.lang.management.MonitorInfo; +import java.lang.management.ThreadInfo; +import java.net.URL; +import java.security.AccessController; +import java.security.PrivilegedAction; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Predicate; +import java.util.stream.Collectors; +import org.apache.kafka.common.config.AbstractConfig; +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigDef.Importance; +import org.apache.kafka.common.config.ConfigDef.Type; +import org.apache.kafka.connect.runtime.WorkerConfig; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TestName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class SynchronizationTest { + +public static final Logger log = LoggerFactory.getLogger(SynchronizationTest.class); + +@Rule +public final TestName testName = new TestName(); + +private String threadPrefix; +private Plugins plugins; +private ThreadPoolExecutor exec; +private Breakpoint dclBreakpoint; +private Breakpoint pclBreakpoint; + +@Before +public void setup() { +TestPlugins.assertAvailable(); +Map pluginProps = Collections.singletonMap( +WorkerConfig.PLUGIN_PATH_CONFIG, +String.join(",", TestPlugins.pluginPath()) +); +threadPrefix = SynchronizationTest.class.getSimpleName() ++ "." + testName.getMethodName() + "-"; +dclBreakpoint = new Breakpoint<>(); +pclBreakpoint = new Breakpoint<>(); +plugins = new Plugins(pluginProps) { +@Override +protected DelegatingClassLoader newDelegatingClassLoader(List paths) { +return AccessController.doPrivileged( +(PrivilegedAction) () -> +new SynchronizedDelegatingClassLoader(paths) +); +} +}; +exec = new ThreadPoolExecutor( +2, +2, +1000L, +TimeUnit.MILLISECONDS, +new LinkedBlockingDeque<>(), +threadFactoryWithNamedThreads(threadPrefix) +); + +} + +@After +public void tearDown() throws InterruptedException { +dclBreakpoint.clear(); +pclBreakpoint.clear(); +exec.shutdown(); +exec.awaitTermination(1L, TimeUnit.SECONDS); +} + +private static class Breakpoint { + +private Predicate predicate; +private CyclicBarrier barrier; + +public synchronized void clear() { +if (barrier != null) { +barrier.reset(); +} +predicate = null; +barrier = null; +} + +public synchronized void set(Predicate predicate) { +clear(); +this.predicate = predicate; +// As soon as the barrier is tripped, the barrier will be reset for the next round. +barrier = new CyclicBarrier(2); +} + +/** + * From a thread under test, await for the test orchestrator to continue execution + * @param obj Object to test with the breakpoint's current predicate + */ +
[GitHub] [kafka] rhauch commented on a change in pull request #8259: KAFKA-7421: Ensure Connect's PluginClassLoader is truly parallel capable and resolve deadlock occurrences
rhauch commented on a change in pull request #8259: URL: https://github.com/apache/kafka/pull/8259#discussion_r633778524 ## File path: connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/SynchronizationTest.java ## @@ -0,0 +1,469 @@ +/* + * 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 org.apache.kafka.connect.runtime.isolation; + +import static org.junit.Assert.fail; + +import java.lang.management.LockInfo; +import java.lang.management.ManagementFactory; +import java.lang.management.MonitorInfo; +import java.lang.management.ThreadInfo; +import java.net.URL; +import java.security.AccessController; +import java.security.PrivilegedAction; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Predicate; +import java.util.stream.Collectors; +import org.apache.kafka.common.config.AbstractConfig; +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigDef.Importance; +import org.apache.kafka.common.config.ConfigDef.Type; +import org.apache.kafka.connect.json.JsonConverter; +import org.apache.kafka.connect.runtime.WorkerConfig; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TestName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class SynchronizationTest { + +public static final Logger log = LoggerFactory.getLogger(SynchronizationTest.class); + +@Rule +public final TestName testName = new TestName(); + +private String threadPrefix; +private Plugins plugins; +private ThreadPoolExecutor exec; +private Breakpoint dclBreakpoint; +private Breakpoint pclBreakpoint; + +@Before +public void setup() { +TestPlugins.assertAvailable(); +Map pluginProps = Collections.singletonMap( +WorkerConfig.PLUGIN_PATH_CONFIG, +String.join(",", TestPlugins.pluginPath()) +); +threadPrefix = SynchronizationTest.class.getSimpleName() ++ "." + testName.getMethodName() + "-"; +dclBreakpoint = new Breakpoint<>(); +pclBreakpoint = new Breakpoint<>(); +plugins = new Plugins(pluginProps) { +@Override +protected DelegatingClassLoader newDelegatingClassLoader(List paths) { +return AccessController.doPrivileged( +(PrivilegedAction) () -> +new SynchronizedDelegatingClassLoader(paths) +); +} +}; +exec = new ThreadPoolExecutor( +2, +2, +1000L, +TimeUnit.MILLISECONDS, +new LinkedBlockingDeque<>(), +threadFactoryWithNamedThreads(threadPrefix) +); + +} + +@After +public void tearDown() throws InterruptedException { +dclBreakpoint.clear(); +pclBreakpoint.clear(); +exec.shutdown(); +exec.awaitTermination(1L, TimeUnit.SECONDS); +} + +private static class Breakpoint { + +private Predicate predicate; +private CyclicBarrier barrier; + +public synchronized void clear() { +if (barrier != null) { +barrier.reset(); +} +predicate = null; +barrier = null; +} + +public synchronized void set(Predicate predicate) { +clear(); +this.predicate = predicate; +// As soon as the barrier is tripped, the barrier will be reset for the next round. +barrier = new CyclicBarrier(2); +} + +/** + * From a thread under test, await for the test orchestrator to continue execution + * @param obj Object to test with