You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "philipnee (via GitHub)" <gi...@apache.org> on 2023/07/27 22:07:53 UTC
[GitHub] [kafka] philipnee opened a new pull request, #14118: KAFKA-14875: Implement wakeup
philipnee opened a new pull request, #14118:
URL: https://github.com/apache/kafka/pull/14118
Continuation of https://github.com/apache/kafka/pull/13490. I closed the original one due to rebase difficulties.
**Summary
Implemented wakeup() mechanism using a WakeupTrigger class to store the pending wakeup item, and when wakeup() is invoked, it checks whether there's an active task or a wakeup task.
- If there's an active task: the task will be completed exceptionally and the atomic reference will be freed up.
- If there an wakedup task, which means wakeup() was invoked before a blocking call was issued. Therefore, the current task will be completed exceptionally immediately.
This PR also addressed minor issues such is:
1. Throwing WakeupException at the right place: As wakeups are thrown by completing an active future exceptionally. The WakeupException is wrapped inside of the ExecutionException.
2. mockConstruction is a thread-lock mock; therefore, we need to free up the reference before completing the test. Otherwise, other tests will continue using the thread-lock mock.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14118:
URL: https://github.com/apache/kafka/pull/14118#issuecomment-1685406078
@junrao - This should be ready for review. There's one failing build but it succeed in a separated build, so I believe it is flaky. Here are the list of flaky tests
```
Build / JDK 11 and Scala 2.13 / testAlterSinkConnectorOffsetsZombieSinkTasks – org.apache.kafka.connect.integration.OffsetsApiIntegrationTest
28s
Build / JDK 11 and Scala 2.13 / testAuthentications(String).quorum=kraft – kafka.api.GroupEndToEndAuthorizationTest
3s
Build / JDK 11 and Scala 2.13 / testNonDefaultConnectionCountLimitAndRateLimit() – kafka.network.ConnectionQuotasTest
10s
Build / JDK 11 and Scala 2.13 / testThreadPoolResize() – kafka.server.DynamicBrokerReconfigurationTest
25s
Build / JDK 11 and Scala 2.13 / [1] Type=ZK, Name=testDualWriteScram, MetadataVersion=3.5-IV2, Security=PLAINTEXT – kafka.zk.ZkMigrationIntegrationTest
32s
Build / JDK 17 and Scala 2.13 / testThreadPoolResize() – kafka.server.DynamicBrokerReconfigurationTest
29s
Build / JDK 17 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
13s
Build / JDK 20 and Scala 2.13 / testThreadPoolResize() – kafka.server.DynamicBrokerReconfigurationTest
33s
Build / JDK 20 and Scala 2.13 / [1] Type=ZK, Name=testDualWriteQuotaAndScram, MetadataVersion=3.5-IV2, Security=PLAINTEXT – kafka.zk.ZkMigrationIntegrationTest
38s
Fixed 192
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1307729523
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active tasks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been woken-up, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
+ } else if (task instanceof ActiveFuture) {
+ ActiveFuture active = (ActiveFuture) task;
+ active.future().completeExceptionally(new WakeupException());
+ return null;
+ } else {
+ return task;
+ }
+ });
+ }
+
+ /*
+ If there is no pending task, set the pending task active.
+ If wakeup was called before setting an active task, the current task will complete exceptionally with
+ WakeupException right
+ away.
+ if there is an active task, throw exception.
+ */
+ public <T> CompletableFuture<T> setActiveTask(final CompletableFuture<T> currentTask) {
+ Objects.requireNonNull(currentTask, "currentTask cannot be null");
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new ActiveFuture(currentTask);
+ } else if (task instanceof WakedupFuture) {
+ currentTask.completeExceptionally(new WakeupException());
+ return null;
+ }
+ // last active state is still active
+ throw new KafkaException("Last active task is still active");
+ });
+ return currentTask;
+ }
+
+ public void clearActiveTask() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return null;
+ } else if (task instanceof ActiveFuture) {
+ return null;
+ }
+ return task;
+ });
+ }
+
+ Wakeupable getPendingTask() {
+ return pendingTask.get();
+ }
+
+ interface Wakeupable { }
+
+ static class ActiveFuture implements Wakeupable {
+ private final CompletableFuture<?> future;
+
+ public ActiveFuture(final CompletableFuture<?> future) {
+ this.future = future;
+ }
+
+ public CompletableFuture<?> future() {
+ return future;
+ }
+ }
+
+ static class WakedupFuture implements Wakeupable { }
Review Comment:
Maybe just WakeupFuture? I guess it is a future for waking up a call.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14118:
URL: https://github.com/apache/kafka/pull/14118#issuecomment-1697747109
hey @junrao - I'm not sure! It does happen from time to time, thanks for rerunning the test.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1305915411
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
Review Comment:
Presumably - 2 spaces?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14118:
URL: https://github.com/apache/kafka/pull/14118#issuecomment-1697932752
Hi @junrao - Just to summarized some of the failing tests:
```
21 tests have failed
There are 19 new tests failing, 2 existing failing and 148 skipped.
New failing - 19
Build / JDK 17 and Scala 2.13 / testReplication() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest
2m 2s
Build / JDK 17 and Scala 2.13 / testThreadPoolResize() – kafka.server.DynamicBrokerReconfigurationTest
32s
Build / JDK 17 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
13s
Build / JDK 17 and Scala 2.13 / initializationError – org.apache.kafka.tiered.storage.integration.OffloadAndConsumeFromLeaderTest
<1s
Build / JDK 17 and Scala 2.13 / testTaskRequestWithOldStartMsGetsUpdated() – org.apache.kafka.trogdor.coordinator.CoordinatorTest
2m 0s
Build / JDK 20 and Scala 2.13 / testFailureToFenceEpoch(String).quorum=kraft – kafka.api.TransactionsTest
58s
Build / JDK 20 and Scala 2.13 / testProducerConsumerOverrideLowerQuota(String).quorum=kraft – kafka.api.UserClientIdQuotaTest
9s
Build / JDK 20 and Scala 2.13 / testNonDefaultConnectionCountLimitAndRateLimit() – kafka.network.ConnectionQuotasTest
10s
Build / JDK 20 and Scala 2.13 / testThreadPoolResize() – kafka.server.DynamicBrokerReconfigurationTest
43s
Build / JDK 20 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
12s
Build / JDK 20 and Scala 2.13 / initializationError – org.apache.kafka.tiered.storage.integration.OffloadAndConsumeFromLeaderTest
<1s
Build / JDK 11 and Scala 2.13 / testThreadPoolResize() – kafka.server.DynamicBrokerReconfigurationTest
37s
Build / JDK 11 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
12s
Build / JDK 11 and Scala 2.13 / initializationError – org.apache.kafka.tiered.storage.integration.OffloadAndConsumeFromLeaderTest
<1s
Build / JDK 8 and Scala 2.12 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
3m 30s
Build / JDK 8 and Scala 2.12 / testRackAwareRangeAssignor() – integration.kafka.server.FetchFromFollowerIntegrationTest
39s
Build / JDK 8 and Scala 2.12 / testThreadPoolResize() – kafka.server.DynamicBrokerReconfigurationTest
35s
Build / JDK 8 and Scala 2.12 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
14s
Build / JDK 8 and Scala 2.12 / initializationError – org.apache.kafka.tiered.storage.integration.OffloadAndConsumeFromLeaderTest
<1s
Existing failures - 2
Build / JDK 17 and Scala 2.13 / testDescribeClusterRequestExcludingClusterAuthorizedOperations(String).quorum=kraft – kafka.server.DescribeClusterRequestTest
2s
Build / JDK 17 and Scala 2.13 / testDescribeClusterRequestIncludingClusterAuthorizedOperations(String).quorum=kraft – kafka.server.DescribeClusterRequestTest
2s
```
Comparing the failing tests against the one before (the difference is really the comment). The two "existing failing" tests, `testDescribeClusterRequestExcludingClusterAuthorizedOperations` didn't really appear in the previous run.
There are a few tests that have been consistently failing across the two builds, but they I believe are unrelated. These are:
- `testThreadPoolResize`
- `testRackAwareRangeAssignor` (server side test)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1308028327
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been wakedup, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
Review Comment:
Sorry @junrao - might have answered the wrong question (sorry, i was too eager to answer...). But yes, poll should clear the pending task when the poll is completed, but currently, the poll isn't functioning at the moment, so I didn't clear the pending task there. For the sake of clarity, do you think I should leave a comment/todo to clarify the intent.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1305857336
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been wakedup, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
Review Comment:
The purpose setting a WakeupFuture when there isn't any active task is because we want the subsequent blocking call the return immediately. Are you wondering if we should maintain this WakeupFuture forever? This is my intend at the moment.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1308011131
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been wakedup, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
Review Comment:
@philipnee : I was asking a slightly different question. Consider that an app calls `Consumer.wakeup` when there is no pending blocking operation. Then the app calls `Consumer.poll` before any blocking operation is called. At this point, a `WakedupFuture` is still left in `WakeupTrigger.pendingTask`. Some point later, a blocking operation is called. It will immediately throw an exception because of `WakedupFuture`, which seems unexpected. So, should `Consumer.poll` clear `WakedupFuture` in `WakeupTrigger.pendingTask`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1279701769
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/OffsetFetchApplicationEvent.java:
##########
@@ -19,25 +19,33 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
-import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
public class OffsetFetchApplicationEvent extends ApplicationEvent {
- final public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
- public final Set<TopicPartition> partitions;
+ private final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+ private final Set<TopicPartition> partitions;
public OffsetFetchApplicationEvent(final Set<TopicPartition> partitions) {
super(Type.FETCH_COMMITTED_OFFSET);
this.partitions = partitions;
this.future = new CompletableFuture<>();
Review Comment:
CompletableApplicationEvent isn't available at the current trunk yet. I think it happens at a different commit.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1306021925
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active tasks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been woken-up, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
+ } else if (task instanceof ActiveFuture) {
+ ActiveFuture active = (ActiveFuture) task;
+ active.future().completeExceptionally(new WakeupException());
+ return null;
+ } else {
+ return task;
+ }
+ });
+ }
+
+ /*
+ If there is no pending task, set the pending task active.
+ If wakeup was called before setting an active task, the current task will complete exceptionally with
+ WakeupException right
+ away.
+ if there is an active task, throw exception.
+ */
+ public <T> CompletableFuture<T> setActiveTask(final CompletableFuture<T> currentTask) {
+ Objects.requireNonNull(currentTask, "currentTask cannot be null");
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new ActiveFuture(currentTask);
+ } else if (task instanceof WakedupFuture) {
+ currentTask.completeExceptionally(new WakeupException());
+ return null;
+ }
+ // last active state is still active
+ throw new KafkaException("Last active task is still active");
+ });
+ return currentTask;
+ }
+
+ public void clearActiveTask() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return null;
+ } else if (task instanceof ActiveFuture) {
+ return null;
+ }
+ return task;
+ });
+ }
+
+ Wakeupable getPendingTask() {
+ return pendingTask.get();
+ }
+
+ interface Wakeupable { }
+
+ static class ActiveFuture implements Wakeupable {
+ private final CompletableFuture<?> future;
+
+ public ActiveFuture(final CompletableFuture<?> future) {
+ this.future = future;
+ }
+
+ public CompletableFuture<?> future() {
+ return future;
+ }
+ }
+
+ static class WakedupFuture implements Wakeupable { }
Review Comment:
Should this be WokenUpFuture?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been wakedup, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
Review Comment:
I was wondering when an application calls `Consumer.wakeup`. If it only calls when it intends to terminate, there is no need to clear WakedupFuture.
However, it seems that an application can call `Consumer.wakeup` to unblock a long `poll` call and then continue to make new `poll` calls. If we don't clear `WakedupFuture`, future blocking calls after the new `poll` call could fail unexpectedly?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1304862712
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
Review Comment:
taks => tasks Also, this method doesn't return anything so it's weird to have "return null here".
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -110,7 +110,7 @@ private boolean process(final OffsetFetchApplicationEvent event) {
Optional<RequestManager> commitRequestManger = registry.get(RequestManager.Type.COMMIT);
CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future = event.future();
if (!commitRequestManger.isPresent()) {
- future.completeExceptionally(new KafkaException("Unable to fetch committed offset because the " +
+ event.future().completeExceptionally(new KafkaException("Unable to fetch committed offset because the " +
Review Comment:
Hmm, why can't we just use `future`?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/WakeupTriggerTest.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.errors.WakeupException;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+public class WakeupTriggerTest {
+ private static long defaultTimeoutMs = 1000;
+ private WakeupTrigger wakeupTrigger;
+
+ @BeforeEach
+ public void setup() {
+ this.wakeupTrigger = new WakeupTrigger();
+ }
+ @Test
Review Comment:
Add a new line above.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been wakedup, do nothing.
Review Comment:
wakedup => wokenUp
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
Review Comment:
Indentation. Ditto below.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been wakedup, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
Review Comment:
Do we ever clear a WakedupFuture from pendingTask if no active task is set afterward?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -469,17 +474,19 @@ public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets) {
@Override
public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets, Duration timeout) {
- CompletableFuture<Void> commitFuture = commit(offsets);
+ CompletableFuture<Void> commitFuture = wakeupTrigger.setActiveTask(commit(offsets));
Review Comment:
Do we need to call `setActiveTask` before the event is added to eventHandler like we did in `committed()`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14118:
URL: https://github.com/apache/kafka/pull/14118#issuecomment-1665733961
I've been seeing stream test failures in the recent commits too!
This PR needs to be rebased against this PR: https://github.com/apache/kafka/pull/14123 to resolve some of the failing committed test.. sorry about that @kirktrue
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kirktrue commented on pull request #14118: KAFKA-14875: Implement wakeup
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #14118:
URL: https://github.com/apache/kafka/pull/14118#issuecomment-1664693959
@philipnee Looks like `PrototypeAsyncConsumerTest.testCommitted()` is failing pretty consistently with this error:
```
org.opentest4j.AssertionFailedError: Unexpected exception thrown: org.apache.kafka.common.errors.TimeoutException: java.util.concurrent.TimeoutException
at app//org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:152)
at app//org.junit.jupiter.api.AssertDoesNotThrow.createAssertionFailedError(AssertDoesNotThrow.java:84)
at app//org.junit.jupiter.api.AssertDoesNotThrow.assertDoesNotThrow(AssertDoesNotThrow.java:75)
at app//org.junit.jupiter.api.AssertDoesNotThrow.assertDoesNotThrow(AssertDoesNotThrow.java:58)
at app//org.junit.jupiter.api.Assertions.assertDoesNotThrow(Assertions.java:3196)
at app//org.apache.kafka.clients.consumer.internals.PrototypeAsyncConsumerTest.testCommitted(PrototypeAsyncConsumerTest.java:163)
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1308022177
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been wakedup, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
Review Comment:
hey @junrao - sorry I misunderstood. In the example you provided, in fact, that is the current behavior. It can be replicated in the integration test like this:
```
@Test
def testWakeup(): Unit = {
val consumer = createConsumer()
val producer = createProducer()
val numRecords = 10000
val startingTimestamp = System.currentTimeMillis()
sendRecords(producer, numRecords, tp, startingTimestamp = startingTimestamp)
consumer.assign(List(tp).asJava)
consumer.wakeup(); // wakeup w/o pending task
assertThrows(classOf[WakeupException], () => consumer.poll(Duration.of(1000, java.time.temporal
.ChronoUnit
.MILLIS))) // throws immediately
println(consumer.poll(Duration.of(1000, java.time.temporal.ChronoUnit.MILLIS))) // prints some records
}
```
In short - consumer.poll shouldn't clear WakeupFuture. In the Javadoc, it also didn't make an exception for the poll call see:
`...the next call to such a method will raise it instead.`
```
Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll. The thread which is blocking in an operation will throw [WakeupException](https://kafka.apache.org/31/javadoc/org/apache/kafka/common/errors/WakeupException.html). If no thread is blocking in a method which can throw [WakeupException](https://kafka.apache.org/31/javadoc/org/apache/kafka/common/errors/WakeupException.html), the next call to such a method will raise it instead.
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1307713180
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been wakedup, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
Review Comment:
Hey Jun - For the blocking calls (syncCommit for example), the future should have been cleared in the finally block. I didn't implement wakeup for the poll method because it is not yet ready at the state of the current trunk. (I'm not sure if i'm answering your question)
Basically - the wakeup future should be cleared in the finally blocks for the blocking APIs, see commitSync. Poll API is still incomplete so I didn't clear the future there.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao closed pull request #14118: KAFKA-14875: Implement wakeup
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao closed pull request #14118: KAFKA-14875: Implement wakeup
URL: https://github.com/apache/kafka/pull/14118
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14118: KAFKA-14875: Implement wakeup
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14118:
URL: https://github.com/apache/kafka/pull/14118#issuecomment-1676661994
Here are the failing tests:
```
Build / JDK 11 and Scala 2.13 / testSimultaneousUpwardAndDownwardDelegating – org.apache.kafka.connect.runtime.isolation.SynchronizationTest
13s
Build / JDK 8 and Scala 2.12 / testBumpTransactionalEpoch(String).quorum=kraft – kafka.api.TransactionsTest
1m 18s
Build / JDK 8 and Scala 2.12 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
13s
Build / JDK 20 and Scala 2.13 / testAlterSinkConnectorOffsetsZombieSinkTasks – org.apache.kafka.connect.integration.OffsetsApiIntegrationTest
55s
Build / JDK 20 and Scala 2.13 / shouldThrottleOldSegments() – kafka.server.ReplicationQuotasTest
9s
Build / JDK 20 and Scala 2.13 / testNoOpActionResponseWithThrottledChannelWhereThrottlingAlreadyDone() – kafka.network.SocketServerTest
<1s
Build / JDK 20 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
12s
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1279692395
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/OffsetFetchApplicationEvent.java:
##########
@@ -19,25 +19,33 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
-import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
public class OffsetFetchApplicationEvent extends ApplicationEvent {
- final public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
- public final Set<TopicPartition> partitions;
+ private final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+ private final Set<TopicPartition> partitions;
public OffsetFetchApplicationEvent(final Set<TopicPartition> partitions) {
super(Type.FETCH_COMMITTED_OFFSET);
this.partitions = partitions;
this.future = new CompletableFuture<>();
Review Comment:
Agree, and it does already. It is just that the changes are not here yet (only in integration branch for now)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kirktrue commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1279620345
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/OffsetFetchApplicationEvent.java:
##########
@@ -19,25 +19,33 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
-import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
public class OffsetFetchApplicationEvent extends ApplicationEvent {
- final public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
- public final Set<TopicPartition> partitions;
+ private final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+ private final Set<TopicPartition> partitions;
public OffsetFetchApplicationEvent(final Set<TopicPartition> partitions) {
super(Type.FETCH_COMMITTED_OFFSET);
this.partitions = partitions;
this.future = new CompletableFuture<>();
Review Comment:
Shouldn't this just extend from `CompletableApplicationEvent` instead?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14118: KAFKA-14875: Implement wakeup
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14118:
URL: https://github.com/apache/kafka/pull/14118#discussion_r1308049575
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/WakeupTrigger.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Ensures blocking APIs can be woken up by the consumer.wakeup().
+ */
+public class WakeupTrigger {
+ private AtomicReference<Wakeupable> pendingTask = new AtomicReference<>(null);
+
+ /*
+ Wakeup a pending task. If there isn't any pending task, return a WakedupFuture, so that the subsequent call
+ would know wakeup was previously called.
+
+ If there are active taks, complete it with WakeupException, then unset pending task (return null here.
+ If the current task has already been wakedup, do nothing.
+ */
+ public void wakeup() {
+ pendingTask.getAndUpdate(task -> {
+ if (task == null) {
+ return new WakedupFuture();
Review Comment:
Thanks for the explanation, @philipnee. Yes, it would be useful to add a TODO comment to make this clear.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on pull request #14118: KAFKA-14875: Implement wakeup
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #14118:
URL: https://github.com/apache/kafka/pull/14118#issuecomment-1697615507
@philipnee : Thanks for the update
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao merged pull request #14118: KAFKA-14875: Implement wakeup
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #14118:
URL: https://github.com/apache/kafka/pull/14118
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kirktrue commented on pull request #14118: KAFKA-14875: Implement wakeup
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #14118:
URL: https://github.com/apache/kafka/pull/14118#issuecomment-1664718076
There are also some streams tests failing that I've not seen before:
```
o.a.k.streams.examples.wordcount.WordCountDemoTest.testCountListOfWords()
o.a.k.streams.examples.wordcount.WordCountDemoTest.testGetStreamsConfig()
o.a.k.streams.examples.wordcount.WordCountDemoTest.testOneWord()
```
I wouldn't think they're related, but let's see if they go away once the main timeout exception issue is resolved.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org