[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1263354270
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/util/SystemTimerReaperTest.java:
##
@@ -0,0 +1,65 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.server.util.timer.SystemTimer;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+
+public class SystemTimerReaperTest {
+private static class FutureTimerTask extends TimerTask {
+CompletableFuture future = new CompletableFuture<>();
+
+FutureTimerTask(long delayMs) {
+super(delayMs);
+}
+
+@Override
+public void run() {
+// We use org.apache.kafka.common.errors.TimeoutException to
differentiate
+// from java.util.concurrent.TimeoutException.
+future.completeExceptionally(new TimeoutException(
+String.format("Future failed to be completed before timeout of
%sMs ms was reached", delayMs)));
+}
+}
+
+private CompletableFuture add(Timer timer, long delayMs) {
+FutureTimerTask task = new FutureTimerTask<>(delayMs);
+timer.add(task);
+return task.future;
+}
+
+@Test
+public void testReaper() throws Exception {
+Timer timer = new SystemTimerReaper("reaper", new
SystemTimer("timer"));
+try {
+CompletableFuture t1 = add(timer, 100L);
+CompletableFuture t2 = add(timer, 200L);
+CompletableFuture t3 = add(timer, 300L);
+TestUtils.assertFutureThrows(t1, TimeoutException.class);
Review Comment:
As I said earlier, the aim of this test is not to verify that tasks are
expired at the right time but rather to verify that tasks are expired at all. I
just wanted to validate that the reaper thread is started and makes progress
here. The SystemTimer already has tests to verify the expiration logic so I
thought that it is not necessary to replicate this here. After all, this class
just introduces the reaper thread.
It seems hard to me to verify that the tasks are expired at the correct time
here without controlling the time and the ticks of the thread. We could try to
measure but that would likely result in a flaky test. I am open to suggestion
here if you have a better idea.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1263349060
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
+private static final long WORK_TIMEOUT_MS = 200L;
Review Comment:
I understand that the system timer and the timing wheel are not easy to
grasp when you don't know them. I also realize that my reply to Jeff
[here](https://github.com/apache/kafka/pull/13991#discussion_r1262824134) was
poorly written. I should have been sharper on this one in order to reduce the
confusion.
Let me try to re-explain how this works in order to reduce the confusion
here. As Jeff said, the time is not necessarily advanced by 200ms every time
the thread ticks. The 200ms is used to wait on the delayed queue which contains
the buckets managed by the timing wheel. This means that it waits maximum 200ms
for a bucket to expire or return. When a bucket expires, the time of the timing
wheel is advanced to the expiration time of the bucket and all the timer tasks
in that bucket are handled. Note that handled here does not mean expired
because the bucket could be from the "overflow wheel". In this case, they may
have to be added back to the current timing wheel or expired.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1263281176
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
+private static final long WORK_TIMEOUT_MS = 200L;
Review Comment:
Haha. Are you saying that I don’t understand what I am doing?
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
+private static final long WORK_TIMEOUT_MS = 200L;
Review Comment:
Haha. Are you saying that I don’t understand what I am doing?
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1263068420
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
+private static final long WORK_TIMEOUT_MS = 200L;
Review Comment:
https://github.com/apache/kafka/blob/trunk/server-common/src/main/java/org/apache/kafka/server/util/timer/TimingWheel.java
has a good javadoc if you want to read more about it.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1263060866
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
+private static final long WORK_TIMEOUT_MS = 200L;
Review Comment:
The important part is that it works like the other reapers that I linked in
another comment.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1263059667
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/util/SystemTimerReaperTest.java:
##
@@ -0,0 +1,65 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.server.util.timer.SystemTimer;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+
+public class SystemTimerReaperTest {
+private static class FutureTimerTask extends TimerTask {
+CompletableFuture future = new CompletableFuture<>();
+
+FutureTimerTask(long delayMs) {
+super(delayMs);
+}
+
+@Override
+public void run() {
+// We use org.apache.kafka.common.errors.TimeoutException to
differentiate
+// from java.util.concurrent.TimeoutException.
+future.completeExceptionally(new TimeoutException(
+String.format("Future failed to be completed before timeout of
%sMs ms was reached", delayMs)));
+}
+}
+
+private CompletableFuture add(Timer timer, long delayMs) {
+FutureTimerTask task = new FutureTimerTask<>(delayMs);
+timer.add(task);
+return task.future;
+}
+
+@Test
+public void testReaper() throws Exception {
+Timer timer = new SystemTimerReaper("reaper", new
SystemTimer("timer"));
+try {
+CompletableFuture t1 = add(timer, 100L);
+CompletableFuture t2 = add(timer, 200L);
+CompletableFuture t3 = add(timer, 300L);
+TestUtils.assertFutureThrows(t1, TimeoutException.class);
Review Comment:
The reaper runs in the background and it advances the time. So the futures
will eventually be completed with the timeout.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1262824134
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
+private static final long WORK_TIMEOUT_MS = 200L;
Review Comment:
i suppose so but i don't know the timing wheel very well.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1262475265
##
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##
@@ -54,18 +58,92 @@ class ConsumerGroupHeartbeatRequestTest(cluster:
ClusterInstance) {
assertEquals(expectedResponse, consumerGroupHeartbeatResponse.data)
}
- @ClusterTest(serverProperties = Array(
+ @ClusterTest(clusterType = Type.KRAFT, serverProperties = Array(
Review Comment:
Correct.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1262474854
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
+private static final long WORK_TIMEOUT_MS = 200L;
Review Comment:
My understanding is that if the clock is advanced by a custom amount, the
items in the `DelayedQueue` will be expired and the next call to `advanceClock`
will process them.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r126246
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
+private static final long WORK_TIMEOUT_MS = 200L;
+
+class Reaper extends ShutdownableThread {
+Reaper(String name) {
+super(name, false);
+}
+
+@Override
+public void doWork() {
+try {
+timer.advanceClock(WORK_TIMEOUT_MS);
+} catch (InterruptedException ex) {
+// Ignore.
+}
+}
+}
+
+private Timer timer;
+private Reaper reaper;
+
+public SystemTimerReaper(String reaperThreadName, Timer timer) {
+this.timer = timer;
+this.reaper = new Reaper(reaperThreadName);
+this.reaper.start();
+}
+
+@Override
+public void add(TimerTask timerTask) {
+timer.add(timerTask);
+}
+
+@Override
+public boolean advanceClock(long timeoutMs) throws InterruptedException {
+return timer.advanceClock(timeoutMs);
+}
+
+@Override
+public int size() {
+return timer.size();
+}
+
+@Override
+public void close() throws Exception {
+reaper.initiateShutdown();
+// Improve shutdown time by waking up any ShutdownableThread(s)
Review Comment:
no others, let me update the comment.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1262462756
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/util/SystemTimerReaperTest.java:
##
@@ -0,0 +1,65 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.server.util.timer.SystemTimer;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+
+public class SystemTimerReaperTest {
+private static class FutureTimerTask extends TimerTask {
+CompletableFuture future = new CompletableFuture<>();
+
+FutureTimerTask(long delayMs) {
+super(delayMs);
+}
+
+@Override
+public void run() {
+// We use org.apache.kafka.common.errors.TimeoutException to
differentiate
+// from java.util.concurrent.TimeoutException.
+future.completeExceptionally(new TimeoutException(
+String.format("Future failed to be completed before timeout of
%sMs ms was reached", delayMs)));
+}
+}
+
+private CompletableFuture add(Timer timer, long delayMs) {
+FutureTimerTask task = new FutureTimerTask<>(delayMs);
+timer.add(task);
+return task.future;
+}
+
+@Test
+public void testReaper() throws Exception {
+Timer timer = new SystemTimerReaper("reaper", new
SystemTimer("timer"));
+try {
+CompletableFuture t1 = add(timer, 100L);
+CompletableFuture t2 = add(timer, 200L);
+CompletableFuture t3 = add(timer, 300L);
+TestUtils.assertFutureThrows(t1, TimeoutException.class);
Review Comment:
We don't. I just added a simple test to ensure that tasks are timed out. I
think that verifying that the correct amount of time has passed is more for the
SystemTimer tests than the reaper.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1262460699
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##
@@ -2360,6 +2360,19 @@ public void testGroupIdsByTopics() {
assertEquals(Collections.emptySet(),
context.groupMetadataManager.groupsSubscribedToTopic("zar"));
}
+@Test
+public void testOnNewMetadataImageWithEmptyDelta() {
Review Comment:
> This is the test for the ofNullable change?
Right.
> Do we also need to check we don't notify any groups?
`testOnNewMetadataImage` already validates that only the correct groups are
notified so I think that it is fine to only validate that `onNewMetadataImage`
does not fail when the delta is empty.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1262458373
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
+private static final long WORK_TIMEOUT_MS = 200L;
+
+class Reaper extends ShutdownableThread {
+Reaper(String name) {
+super(name, false);
+}
+
+@Override
+public void doWork() {
+try {
+timer.advanceClock(WORK_TIMEOUT_MS);
Review Comment:
To be honest, I don't know. I just reused the magic number that we already
used:
*
https://github.com/apache/kafka/blob/8d24716f27b307da79a819487aefb8dec79b4ca8/core/src/main/scala/kafka/raft/TimingWheelExpirationService.scala#L56
*
https://github.com/apache/kafka/blob/8d24716f27b307da79a819487aefb8dec79b4ca8/core/src/main/scala/kafka/server/DelayedOperation.scala#L444
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1262456201
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/SystemTimerReaper.java:
##
@@ -0,0 +1,81 @@
+/*
+ * 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.coordinator.group.util;
+
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.server.util.timer.TimerTask;
+
+/**
+ * SystemTimerReaper wraps a {@link Timer} and starts a reaper thread
+ * to expire the tasks in the {@link Timer}.
+ */
+public class SystemTimerReaper implements Timer {
Review Comment:
Correct. The system timer does not actually have a thread to expire tasks.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1262455254
##
core/src/main/scala/kafka/server/KafkaConfig.scala:
##
@@ -175,7 +176,7 @@ object Defaults {
val ConsumerGroupMinHeartbeatIntervalMs = 5000
val ConsumerGroupMaxHeartbeatIntervalMs = 15000
val ConsumerGroupMaxSize = Int.MaxValue
- val ConsumerGroupAssignors = ""
+ val ConsumerGroupAssignors = List(classOf[RangeAssignor].getName).asJava
Review Comment:
The KIP says that the default should be `uniform, range`. We don't have the
uniform yet so I have already put the range one here.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1260620995
##
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##
@@ -54,18 +58,92 @@ class ConsumerGroupHeartbeatRequestTest(cluster:
ClusterInstance) {
assertEquals(expectedResponse, consumerGroupHeartbeatResponse.data)
}
- @ClusterTest(serverProperties = Array(
+ @ClusterTest(clusterType = Type.KRAFT, serverProperties = Array(
Review Comment:
Yeah, we will. We have to do two things: 1) parameterise existing
integration/system tests to run with the new group coordinator. i hope that
this will cover a bug chunk of the new coordinator; and 2) extend
integration/system tests where needed.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13991: KAFKA-14462; [23/23] Wire GroupCoordinatorService in BrokerServer
dajac commented on code in PR #13991:
URL: https://github.com/apache/kafka/pull/13991#discussion_r1260620995
##
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##
@@ -54,18 +58,92 @@ class ConsumerGroupHeartbeatRequestTest(cluster:
ClusterInstance) {
assertEquals(expectedResponse, consumerGroupHeartbeatResponse.data)
}
- @ClusterTest(serverProperties = Array(
+ @ClusterTest(clusterType = Type.KRAFT, serverProperties = Array(
Review Comment:
Yeah, we will. We have to do two things: 1) parameterise existing
integration/system tests to run with the new group coordinator; and 2) extend
integration/system tests where needed.
--
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
