gharris1727 commented on code in PR #12290:
URL: https://github.com/apache/kafka/pull/12290#discussion_r1424413335
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java:
##########
@@ -350,13 +353,16 @@ private void assertRequestTimesOut(String
requestDescription, ThrowingRunnable r
}
private static class Block {
Review Comment:
can you make this public to allow OffsetsApiIntegrationTest to use the latch?
and do you think that maybe these connectors should be moved out of this
test to a common reusable class?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java:
##########
@@ -368,31 +374,54 @@ private static ConfigDef config() {
);
}
+ /**
+ * {@link CountDownLatch#await() Wait} for the connector/task to reach
the point in its lifecycle where
+ * it will block.
+ */
public static void waitForBlock() throws InterruptedException,
TimeoutException {
+ CountDownLatch awaitBlockLatch;
synchronized (Block.class) {
- if (blockLatch == null) {
- throw new IllegalArgumentException("No connector has been
created yet");
- }
+ awaitBlockLatch = Block.awaitBlockLatch;
+ }
+
+ if (awaitBlockLatch == null) {
+ throw new IllegalArgumentException("No connector has been
created yet");
Review Comment:
Is this an opportunity for a flaky failure, if the test thread advances
before the connector is created. It seems very rare, I don't see any instances
on the Gradle dashboard.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java:
##########
@@ -368,31 +374,54 @@ private static ConfigDef config() {
);
}
+ /**
+ * {@link CountDownLatch#await() Wait} for the connector/task to reach
the point in its lifecycle where
+ * it will block.
+ */
public static void waitForBlock() throws InterruptedException,
TimeoutException {
+ CountDownLatch awaitBlockLatch;
synchronized (Block.class) {
- if (blockLatch == null) {
- throw new IllegalArgumentException("No connector has been
created yet");
- }
+ awaitBlockLatch = Block.awaitBlockLatch;
+ }
+
+ if (awaitBlockLatch == null) {
+ throw new IllegalArgumentException("No connector has been
created yet");
}
log.debug("Waiting for connector to block");
- if (!blockLatch.await(CONNECTOR_BLOCK_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
+ if (!awaitBlockLatch.await(CONNECTOR_BLOCK_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
throw new TimeoutException("Timed out waiting for connector to
block.");
Review Comment:
Since scanning creates connector instances, and validation caches the
connector instance, how do you ensure that the right awaitBlockLatch is being
waited on here?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java:
##########
@@ -139,7 +141,8 @@ public void setup() throws Exception {
public void close() {
// stop all Connect, Kafka and Zk threads.
connect.stop();
- Block.resetBlockLatch();
+ // unblock everything so that we don't leak threads after each test run
+ Block.reset();
Review Comment:
WDYT about resetting before stopping the workers, to allow a normal shutdown
to happen?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
