You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "cmccabe (via GitHub)" <gi...@apache.org> on 2023/05/24 23:36:49 UTC
[GitHub] [kafka] cmccabe opened a new pull request, #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
cmccabe opened a new pull request, #13759:
URL: https://github.com/apache/kafka/pull/13759
When the active KRaft controller is overloaded, it will not be able to process broker heartbeat
requests. Instead, they will be timed out. When using the default configuration, this will happen
if the time needed to process a broker heartbeat climbs above a second for a sustained period.
This, in turn, could lead to brokers being improperly fenced when they are still alive.
This PR creates a new state, the overload state, which the active controller will enter when it
fails to process more than 3 heartbeats in a given 5 minute period. While in overload state, no
brokers will be fenced due to missed heartbeats (because we don't know if the heartbeats were
missed because of load, or another reason).
Entering overload state increments the metadata error metric on the controller, which will notify
the operator about the problem. We also log a message when we notice that we are out of overload
state.
BrokerHeartbeatManager now stores the data about how many heartbeats have been missed. I added a
builder to make it easier to add more constructor parameters.
WindowedEventCounter is a new generic class which counts the number of times it has seen a specific
event within a certain time period.
In addition to the changes above, in QuorumController, we now increment the metadata fault metric
if we get an unexpected exception while processing an event. (In other words, an exception that
doesn't map cleanly to a known return code.) This should help catch NullPointerExceptions and
similar, if any surface again.
--
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] cmccabe commented on pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on PR #13759:
URL: https://github.com/apache/kafka/pull/13759#issuecomment-1568871804
Hi all,
I've updated the approach in the PR. The new approach simply adds basic heartbeat handling in the timeout case. We can't do everything we normally can here. Most notably, anything involving heartbeats causing a state change like fenced -> unfenced or entering controlled shutdown is off limits. But it's sufficient to prevent the congestion collapse scenario, and it's pretty simple.
I will also create a KIP to add a new metric for timed out broker heartbeats, since I think that's very worthwhile to do. We can do that in a follow-on PR. This is something we'll want to track pretty closely. If heartbeats start timing out that indicates a critical performance problem that needs to be addressed by the operator.
--
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] cmccabe commented on a diff in pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1211991966
##########
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java:
##########
@@ -1419,6 +1421,15 @@ ControllerResult<BrokerHeartbeatReply> processBrokerHeartbeat(
return ControllerResult.of(records, reply);
}
+ void processExpiredBrokerHeartbeat(BrokerHeartbeatRequestData request) {
+ int brokerId = request.brokerId();
+ clusterControl.checkBrokerEpoch(brokerId, request.brokerEpoch());
+ clusterControl.heartbeatManager().touch(brokerId,
+ clusterControl.brokerRegistrations().get(brokerId).fenced(),
+ request.currentMetadataOffset());
+ log.error("processExpiredBrokerHeartbeat: timed out heartbeat from broker {}.", brokerId);
Review Comment:
I changed it to "controller event queue overloaded. timed out heartbeat from broker {}"
--
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] cmccabe commented on a diff in pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1210858865
##########
metadata/src/main/java/org/apache/kafka/controller/errors/ControllerExceptions.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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.controller.errors;
+
+import org.apache.kafka.common.errors.TimeoutException;
+
+import java.util.concurrent.ExecutionException;
+
+
+public class ControllerExceptions {
+ /**
+ * Check if an exception is a normal timeout exception.
+ *
+ * @param exception The exception to check.
+ * @return True if the exception is a normal timeout exception and NOT a timeout
+ * exception generated by the controller shutting down.
+ */
+ public static boolean isNormalTimeoutException(Throwable exception) {
+ if (exception == null) return false;
+ if (exception instanceof ExecutionException) {
+ exception = exception.getCause();
+ if (exception == null) return false;
+ }
+ if (!(exception instanceof TimeoutException)) return false;
+ if (exception.getMessage() != null &&
+ exception.getMessage().equals("The controller is shutting down.")) {
+ return false;
+ }
+ return true;
Review Comment:
I removed the check for the exception text. (It wasn't actually doing anything here in any case!) This is something I'll revisit in a follow-on PR (there are a few corner cases around error handling and queue shutdown to fix up)
--
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] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1207013139
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
Thinking about this some more, I believe the key requirement is to accurately understand when we do not have a correct view of the cluster. This allows us to handle 2 important cases: not fencing a broker if its session times out but we think we could have missed enough heartbeats to make the decision to fence the wrong decision; and fencing a broker if its session times out whenever we think we have accurate information. I think I addressed the second part (detect and fence a crashed broker as quick as possible) in the comment above: I believe we have accurate information if we see a contiguous series of N successfully-processed heartbeats with no intervening timed-out heartbeats where N is perhaps the cluster size. For the first part (don't fence if we think the decision to do so could be wrong) assume the broker session is 18 seconds and the heartbeat interval is 2 seconds. That means we would need to miss 9 heartbeats for a broker in order to incorrectly fence it. Maybe
we keep track of the last time we had enough contiguous successful heartbeats (which, if we aren't missing any would always be very recent). But then as soon as we miss one we increment the missed count and reset the contiguous count to 0. When we successfully process a heartbeat we increment the contiguous count and, if it reaches the necessary threshold N (which is on the order of the cluster size) we reset the missed count to 0. We can fence brokers only while the missed count is less than the session/interval ratio (i.e. 18/2 = 9 by default).
We can tweak this to be a bit more conservative. Maybe we need N (the number of contiguous heartbeats seen to assure us we have good visibility) to be 1.5 or 2 times the broker count instead of just the broker count. Maybe the missed count only has to exceed half the session/interval ratio (so only missing 5 heartbeats without seeing N successfully-processed ones in a row would prevent fencing instead of 9 by default).
WDYT?
--
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] rondagostino commented on pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on PR #13759:
URL: https://github.com/apache/kafka/pull/13759#issuecomment-1568668239
> The 3 missed heartbeats in your example would result in 3 brokers being fenced by the time the overload state is entered? Or did I misunderstand? And what happens next?
Using the default values of broker session timeout = 9 seconds and broker heartbeat interval = 2 seconds, we would need to miss at least 4 heartbeats before any single broker could potentially lose its session. And with 3 brokers there would by 3 heartbeats arriving every 2 seconds, so after 8 seconds we would typically receive 3 * 4 = 12 heartbeats. If we just missed 3 of them it is conceivable that they would all have been for the same broker.
But note that Colin states above `I think we may not need the overload state I originally wanted to introduce at all. We can simply do some basic processing on the heartbeat when we time it out. Specifically, we can update the "last seen time" of the broker."`. I believe what he is saying is that when we see a heartbeat request in the event queue and it is old enough such that we "missed" it, we can still say that the broker that sent it was alive and able to contact us at that time (whenever it was -- it could have been several seconds ago by the time we see it). So while we can't respond to it -- the broker has timed-out the request and is no longer waiting for the response -- we can still note that the broker was alive at that point. At first glance this seems like a lot better approach than introducing this "overload" state -- as the discussion shows, such a state isn't intuitive to reason about. And I see no reason why we can't take the approach of counting a heartbeat th
at we don't respond to as indicating the broker was alive...
--
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] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1207013139
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
Thinking about this some more, I believe the key requirement is to accurately understand when we do not have a correct view of the cluster. This allows us to handle 2 important cases: not fencing a broker if its session times out but we think we could have missed enough heartbeats to make the decision to fence the wrong decision; and fencing a broker if its session times out whenever we think we have accurate information. I think I addressed the second part (detect and fence a crashed broker as quick as possible) in the comment above: I believe we have accurate information if we see a contiguous series of N successfully-processed heartbeats with no intervening timed-out heartbeats where N is perhaps the cluster size. For the first part (don't fence if we think the decision to do so could be wrong) assume the broker session is 18 seconds and the heartbeat interval is 2 seconds. That means we would need to miss 9 heartbeats for a broker in order to incorrectly fence it. Maybe
we keep track of the last time we had enough contiguous successful heartbeats (which, if we aren't missing any would always be very recent). But then as soon as we miss one we increment the missed count and reset the contiguous count to 0. When we successfully process a heartbeat we increment the contiguous count and, if it reaches the necessary threshold N (which is on the order of the cluster size) we reset the missed count to 0. We can fence brokers only when the missed count approaches the session/interval ratio (i.e. 18/2 = 9 by default).
We can tweak this to be a bit more conservative. Maybe we need N (the number of contiguous heartbeats seen to assure us we have good visibility) to be 1.5 or 2 times the broker count instead of just the broker count. Maybe the missed count only has to exceed half the session/interval ratio (so only missing 5 heartbeats without seeing N successfully-processed ones in a row would prevent fencing instead of 9 by default).
WDYT?
--
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] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1207013139
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
Thinking about this some more, I believe the key requirement is to accurately understand when we do not have a correct view of the cluster. This allows us to handle 2 important cases: not fencing a broker if its session times out but we think we could have missed enough heartbeats to make the decision to fence the wrong decision; and fencing a broker if its session times out whenever we think we have accurate information. I think I addressed the second part (detect and fence a crashed broker as quick as possible) in the comment above: I believe we have accurate information if we see a contiguous series of N successfully-processed heartbeats with no intervening timed-out heartbeats where N is perhaps the cluster size. For the first part (don't fence if we think the decision to do so could be wrong) assume the broker session is 18 seconds and the heartbeat interval is 2 seconds. That means we would need to miss 9 heartbeats for a broker in order to incorrectly fence it. Maybe
we keep track of the last time we had enough contiguous successful heartbeats (which, if we aren't missing any would always be very recent). But then as soon as we miss one we increment the missed count and reset the contiguous count to 0. When we successfully process a heartbeat we increment the contiguous count and, if it reaches the necessary threshold N (which is on the order of the cluster size) we reset the missed count to 0. We can fence brokers only while the missed count is less than the session/interval ratio (i.e. 18/2 = 9 by default).
We can tweak this to be a bit more conservative. Maybe we need N (the number of contiguous heartbeats seen to assure us we have good visibility) to be 1.5 or 2 times the broker count instead of just the broker count. Maybe the missed count only has to exceed half the session/interval ratio (so only missing 5 heartbeats without seeing N successfully-processed ones in a row instead of 9 by default) to prevent fencing.
WDYT?
--
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] ijuma commented on a diff in pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1212083820
##########
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java:
##########
@@ -1419,6 +1421,23 @@ ControllerResult<BrokerHeartbeatReply> processBrokerHeartbeat(
return ControllerResult.of(records, reply);
}
+ /**
+ * Process a broker heartbeat which has been sitting on the queue for too long, and has
+ * expired. With default settings, this would happen after 1 second. We process expired
+ * heartbeats by updating the lastSeenNs of the broker, so that the broker won't get fenced
+ * incorrectly. However, we don't perform any state changes that we normally would, such as
+ * unfencing a fenced broker, etc.
+ */
Review Comment:
Thanks for this comment, it's very helpful.
--
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] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1205924251
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -613,4 +733,32 @@ BrokerControlStates calculateNextBrokerState(int brokerId,
return new BrokerControlStates(currentState, SHUTDOWN_NOW);
}
}
+
+ int timedOutHeartbeats(long nowNs) {
Review Comment:
`private`? It seems all methods are currently package-private. Should any of them be `private` instead?
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -55,6 +59,62 @@
* active controller does not know when the last heartbeats were received from each.
*/
public class BrokerHeartbeatManager {
+ static class Builder {
+ private LogContext logContext = null;
+ private Time time = Time.SYSTEM;
+ private long sessionTimeoutNs = 8000;
Review Comment:
Should probably default to `ClusterControlManager.DEFAULT_SESSION_TIMEOUT_NS` (which is 18_000_000_000, it appears you put in a milliseconds value, and the incorrect one at that).
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -613,4 +733,32 @@ BrokerControlStates calculateNextBrokerState(int brokerId,
return new BrokerControlStates(currentState, SHUTDOWN_NOW);
}
}
+
+ int timedOutHeartbeats(long nowNs) {
+ return timedOutHeartbeatCounter.count(nowNs);
+ }
+
+ boolean overloaded(long nowNs) {
+ return timedOutHeartbeats(nowNs) > timedOutHeartbeatOverloadThreshold;
+ }
+
+ void handleBrokerHeartbeatTimeout() {
+ long nowNs = time.nanoseconds();
+ boolean alreadyOverloaded = overloaded(nowNs);
+ timedOutHeartbeatCounter.addEvent(nowNs);
+ if ((!alreadyOverloaded) && overloaded(nowNs)) {
+ nonFatalFaultHandler.handleFault(String.format("handleBrokerHeartbeatTimeout: the " +
+ "active controller missed %d heartbeats in the last %d ms and is now " +
+ "overloaded.", timedOutHeartbeatOverloadThreshold,
+ NANOSECONDS.toMillis(timedOutHeartbeatCounter.window())));
+ }
+ }
+
+ boolean maybeLogOverloadExit(long nowNs) {
+ if (!needToLogOverloadExit) return false;
+ if (overloaded(time.nanoseconds())) return false;
+ log.error("The active controller is no longer overloaded.");
Review Comment:
Should this be at the `ERROR` level?
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -613,4 +733,32 @@ BrokerControlStates calculateNextBrokerState(int brokerId,
return new BrokerControlStates(currentState, SHUTDOWN_NOW);
}
}
+
+ int timedOutHeartbeats(long nowNs) {
+ return timedOutHeartbeatCounter.count(nowNs);
+ }
+
+ boolean overloaded(long nowNs) {
+ return timedOutHeartbeats(nowNs) > timedOutHeartbeatOverloadThreshold;
+ }
+
+ void handleBrokerHeartbeatTimeout() {
+ long nowNs = time.nanoseconds();
+ boolean alreadyOverloaded = overloaded(nowNs);
+ timedOutHeartbeatCounter.addEvent(nowNs);
+ if ((!alreadyOverloaded) && overloaded(nowNs)) {
+ nonFatalFaultHandler.handleFault(String.format("handleBrokerHeartbeatTimeout: the " +
Review Comment:
Looks like we need to set `needToLogOverloadExit = true` here.
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
Whatever we set this to (and currently we always take this default value) will determine how long it would take for a controller to leave the overloaded state, right? So if we see a series of missed heartbeats that puts us into the overloaded state, and then we don't see any more missed heartbeats, it would take this amount of time to leave that overloaded state?
Consider the following case. We are overloaded. And then some broker crashes. How long would we want it to take before we move leadership away for any partitions led by that suddenly-dead broker? Ideally the session timeout, of course, but that won't happen if we are overloaded -- will it take 5 minutes as currently coded? That seems like a long time.
I think the only thing we can say is that we should move leadership away as soon as we have a correct view of the cluster. The question then becomes: how do we know when we have good information?
A fixed amount of time like this with no missed heartbeats -- that seems too simple. We likely have a reasonable view of the cluster if we see enough heartbeats without missing any such that it is likely that all brokers have had a chance to communicate with us. So it seems like the broker heartbeat interval should factor into it somehow.
Let's say the broker heartbeat interval is 2 seconds. Right now we consider a heartbeat missed if it is more than half the heartbeat interval (i.e. 1 second) old. Let's say there are 60 brokers in the cluster. If we see something on the order of 60 heartbeats in row without having missing any, then we probably have a reasonable view.
So maybe it isn't having missed no more than N heartbeats in some fixed time window. Maybe we define "not overloaded" as having seen some string of heartbeats uninterrupted without seeing a missed one? So every time we miss a heartbeat we go into the "overloaded" state and we reset a counter of contiguous successfully processed heartbeats to 0. Whenever we see a heartbeat and process it in time we increase that counter. We are overloaded when the `contiguousHeartbeatsProcessedInTime` counter is equal to or exceeds the broker count.
I'm not sure I ave it right, but I think this is a discussion worth pursuing.
##########
server-common/src/main/java/org/apache/kafka/server/metrics/WindowedEventCounter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.server.metrics;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+/**
+ * Counts the number of events that happen within a given time window. Not thread safe.
+ */
+public final class WindowedEventCounter {
+ private final long[] eventTimes;
+
+ private final long window;
+
+ private int startIndex;
+
+ private int endIndex;
+
+ public WindowedEventCounter(
+ int maxCount,
+ long window
+ ) {
Review Comment:
Javadoc would help. A rename to `windowNs` or something like that would be helpful to clearly indicate it is nanoseconds.
##########
server-common/src/main/java/org/apache/kafka/server/metrics/WindowedEventCounter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.server.metrics;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+/**
+ * Counts the number of events that happen within a given time window. Not thread safe.
+ */
+public final class WindowedEventCounter {
+ private final long[] eventTimes;
+
+ private final long window;
+
+ private int startIndex;
+
+ private int endIndex;
+
+ public WindowedEventCounter(
+ int maxCount,
+ long window
+ ) {
+ if (maxCount <= 0) {
+ throw new IllegalArgumentException("maxCount must be positive");
+ }
+ if (window <= 0) {
+ throw new IllegalArgumentException("window must be positive");
+ }
+ this.eventTimes = new long[maxCount + 1];
+ this.window = window;
+ this.startIndex = 0;
+ this.endIndex = 0;
+ }
+
+ public int count(long now) {
Review Comment:
I think the parameter rename `s/now/nowNs` (and everywhere below) would add clarity.
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -55,6 +59,62 @@
* active controller does not know when the last heartbeats were received from each.
*/
public class BrokerHeartbeatManager {
+ static class Builder {
+ private LogContext logContext = null;
+ private Time time = Time.SYSTEM;
+ private long sessionTimeoutNs = 8000;
+ private FaultHandler nonFatalFaultHandler = (message, cause) -> new RuntimeException(message, cause);
+ private int timedOutHeartbeatCountMax = DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX;
+ private long timedOutHeartbeatCountWindowNs = DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS;
+ private int timedOutHeartbeatOverloadThreshold = DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD;
+
+ Builder setLogContext(LogContext logContext) {
+ this.logContext = logContext;
+ return this;
+ }
+
+ Builder setTime(Time time) {
+ this.time = time;
+ return this;
+ }
+
+ Builder setSessionTimeoutNs(long sessionTimeoutNs) {
+ this.sessionTimeoutNs = sessionTimeoutNs;
+ return this;
+ }
+
+ Builder setNonFatalFaultHandler(FaultHandler nonFatalFaultHandler) {
+ this.nonFatalFaultHandler = nonFatalFaultHandler;
+ return this;
+ }
+
+ Builder setTimedOutHeartbeatCountMax(int timedOutHeartbeatCountMax) {
+ this.timedOutHeartbeatCountMax = timedOutHeartbeatCountMax;
+ return this;
+ }
+
+ Builder setTimedOutHeartbeatCountMs(long timedOutHeartbeatCountWindowNs) {
+ this.timedOutHeartbeatCountWindowNs = timedOutHeartbeatCountWindowNs;
+ return this;
Review Comment:
Method name: `s/Ms/WindowNs/`
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -402,6 +509,19 @@ void touch(int brokerId, boolean fenced, long metadataOffset) {
}
}
+ /**
+ * Update the lastContactNs of a broker we already know about.
+ *
+ * @param brokerId The broker ID.
+ */
+ void touchExisting(int brokerId) {
Review Comment:
Since this is invoked in the overload state to indicate that we are giving a new grace period, should we rename `BrokerHeartbeatState.lastContactNs` and change its Javadoc? I think the semantics would be reasonably reflected with a name like `lastSessionGracePeriodStartNs` or something similar.
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -2085,7 +2094,7 @@ public CompletableFuture<BrokerHeartbeatReply> processBrokerHeartbeat(
ControllerRequestContext context,
BrokerHeartbeatRequestData request
) {
- return appendWriteEvent("processBrokerHeartbeat", context.deadlineNs(),
+ CompletableFuture<BrokerHeartbeatReply> future = appendWriteEvent("processBrokerHeartbeat", context.deadlineNs(),
Review Comment:
It might be helpful to append the broker ID to the name here so that we can see in the logs which broker heartbeat was missed.
--
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] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1206028552
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
Whatever we set this to (and currently we always take this default value) will determine how long it would take for a controller to leave the overloaded state, right? So if we see a series of missed heartbeats that puts us into the overloaded state, and then we don't see any more missed heartbeats, it would take this amount of time to leave that overloaded state?
Consider the following case. We are overloaded. And then some broker crashes. How long would we want it to take before we move leadership away for any partitions led by that suddenly-dead broker? Ideally the session timeout, of course, but that won't happen if we are overloaded -- will it take 5 minutes as currently coded? That seems like a long time.
I think the only thing we can say is that we should move leadership away as soon as we have a correct view of the cluster. The question then becomes: how do we know when we have good information?
A fixed amount of time like this with no missed heartbeats -- that seems too simple. We likely have a reasonable view of the cluster if we see enough heartbeats without missing any such that it is likely that all brokers have had a chance to communicate with us. So it seems like maybe the broker heartbeat interval should factor into it somehow?
Let's say the broker heartbeat interval is 2 seconds. Right now we consider a heartbeat missed if it is more than half the heartbeat interval (i.e. 1 second) old. Let's say there are 60 brokers in the cluster. If we see something on the order of 60 heartbeats in row without having missing any, then we probably have a reasonable view.
So maybe it isn't having missed no more than N heartbeats in some fixed time window. Maybe we define "not overloaded" as having seen some string of heartbeats uninterrupted without seeing a missed one? So every time we miss a heartbeat we go into the "overloaded" state and we reset a counter of contiguous successfully processed heartbeats to 0. Whenever we see a heartbeat and process it in time we increase that counter. We are no longer overloaded when the `contiguousHeartbeatsProcessedInTime` counter is equal to or exceeds the broker count.
I'm not sure I ave it right, but I think this is a discussion worth pursuing.
--
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] ijuma commented on a diff in pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1211352189
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -74,7 +74,7 @@
* brokers being fenced or unfenced, and broker feature versions.
*/
public class ClusterControlManager {
- final static long DEFAULT_SESSION_TIMEOUT_NS = NANOSECONDS.convert(18, TimeUnit.SECONDS);
+ final static long DEFAULT_SESSION_TIMEOUT_NS = NANOSECONDS.convert(9, TimeUnit.SECONDS);
Review Comment:
Why was this default value changed? It wasn't clear from the PR description.
--
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] cmccabe commented on a diff in pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1210806096
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -1106,6 +1107,34 @@ class KRaftClusterTest {
cluster.close()
}
}
+
+ @Test
+ def testTimedOutHeartbeats(): Unit = {
+ val cluster = new KafkaClusterTestKit.Builder(
+ new TestKitNodes.Builder().
+ setNumBrokerNodes(3).
+ setNumControllerNodes(1).build()).
+ setConfigProp(KafkaConfig.BrokerHeartbeatIntervalMsProp, 10.toString).
+ setConfigProp(KafkaConfig.BrokerSessionTimeoutMsProp, 1000.toString).
+ build()
+ try {
+ cluster.format()
+ cluster.startup()
+ val controller = cluster.controllers().values().iterator().next()
+ controller.controller.waitForReadyBrokers(3).get()
+ TestUtils.retry(60000) {
+ val latch = controller.controller.asInstanceOf[QuorumController].pause()
+ Thread.sleep(1001)
+ latch.countDown()
+ controller.controller.asInstanceOf[QuorumController].pause().countDown()
Review Comment:
Removed
--
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] ijuma commented on a diff in pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1211357274
##########
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java:
##########
@@ -1419,6 +1421,15 @@ ControllerResult<BrokerHeartbeatReply> processBrokerHeartbeat(
return ControllerResult.of(records, reply);
}
+ void processExpiredBrokerHeartbeat(BrokerHeartbeatRequestData request) {
+ int brokerId = request.brokerId();
+ clusterControl.checkBrokerEpoch(brokerId, request.brokerEpoch());
+ clusterControl.heartbeatManager().touch(brokerId,
+ clusterControl.brokerRegistrations().get(brokerId).fenced(),
+ request.currentMetadataOffset());
+ log.error("processExpiredBrokerHeartbeat: timed out heartbeat from broker {}.", brokerId);
Review Comment:
What is the operator meant to do when they see this error message? Perhaps we can explain the implications in the log message?
--
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] cmccabe commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1207395237
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
I agree with you that 5 minutes is way too long to "mute" broker removals. That was a miss in the orignial PR. (As a side note, broker session is not 18 seconds. It is 9 seconds.)
I thought about this more and I think we may not need the overload state I originally wanted to introduce at all. We can simply do some basic processing on the heartbeat when we time it out. Specifically, we can update the "last seen time" of the broker. This will avoid the "congestion meltdown" behavior without introducing a new state.
--
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] cmccabe commented on a diff in pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1210858865
##########
metadata/src/main/java/org/apache/kafka/controller/errors/ControllerExceptions.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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.controller.errors;
+
+import org.apache.kafka.common.errors.TimeoutException;
+
+import java.util.concurrent.ExecutionException;
+
+
+public class ControllerExceptions {
+ /**
+ * Check if an exception is a normal timeout exception.
+ *
+ * @param exception The exception to check.
+ * @return True if the exception is a normal timeout exception and NOT a timeout
+ * exception generated by the controller shutting down.
+ */
+ public static boolean isNormalTimeoutException(Throwable exception) {
+ if (exception == null) return false;
+ if (exception instanceof ExecutionException) {
+ exception = exception.getCause();
+ if (exception == null) return false;
+ }
+ if (!(exception instanceof TimeoutException)) return false;
+ if (exception.getMessage() != null &&
+ exception.getMessage().equals("The controller is shutting down.")) {
+ return false;
+ }
+ return true;
Review Comment:
I removed the check for the exception text. This is something I'll revisit in a follow-on PR (there are a few corner cases around error handling and queue shutdown to fix up)
--
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] cmccabe commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1207382044
##########
server-common/src/main/java/org/apache/kafka/server/metrics/WindowedEventCounter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.server.metrics;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+/**
+ * Counts the number of events that happen within a given time window. Not thread safe.
+ */
+public final class WindowedEventCounter {
+ private final long[] eventTimes;
+
+ private final long window;
+
+ private int startIndex;
+
+ private int endIndex;
+
+ public WindowedEventCounter(
+ int maxCount,
+ long window
+ ) {
Review Comment:
You can use any units you'd like here. For example it would be possible to use either ms or ns.
--
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] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1207013139
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
Thinking about this some more, I believe the key requirement is to accurately understand when we do not have a correct view of the cluster. This allows us to handle 2 important cases: not fencing a broker if its session times out but we think we could have missed enough heartbeats to make the decision to fence the wrong decision; and fencing a broker if its session times out whenever we think we have accurate information. I think I addressed the second part (detect and fence a crashed broker as quick as possible) in the comment above: I believe we have accurate information if we see a contiguous series of N successfully-processed heartbeats with no intervening timed-out heartbeats where N is perhaps the cluster size. For the first part (don't fence if we think the decision to do so could be wrong) assume the broker session is 18 seconds and the heartbeat interval is 2 seconds. That means we would need to miss 9 heartbeats for a broker in order to incorrectly fence it. Maybe
we keep track of the number of contiguous successful heartbeats with no intervening misses (which, if we aren't missing any would always be a very high number). But then as soon as we miss one we increment the missed count and reset the contiguous count to 0. When we successfully process a heartbeat we increment the contiguous count and, if it reaches the necessary threshold N (which is on the order of the cluster size) we reset the missed count to 0. We can fence brokers only while the missed count is less than the session/interval ratio (i.e. 18/2 = 9 by default).
We can tweak this to be a bit more conservative. Maybe we need N (the number of contiguous heartbeats seen to assure us we have good visibility) to be 1.5 or 2 times the broker count instead of just the broker count. Maybe the missed count only has to exceed half the session/interval ratio (so only missing 5 heartbeats without seeing N successfully-processed ones in a row instead of 9 by default) to prevent fencing.
WDYT?
--
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] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1210468300
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
> As a side note, broker session is not 18 seconds. It is 9 seconds.
I was confused by the existence of this, which sets 18 seconds: https://github.com/apache/kafka/blob/d557f7997359daa5853d446d9093995577540de2/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java#L77
But the above always get overridden by the value of `broker.session.timeout.ms` in the config, which does default to 9 seconds (https://github.com/apache/kafka/blob/d557f7997359daa5853d446d9093995577540de2/core/src/main/scala/kafka/server/KafkaConfig.scala#L83).
So yes, it is 9 seconds as you state above. I wonder if the default in `ClusterControlManager.java` should be changed to match the config's default.
--
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] cmccabe commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1210644585
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
I'll update 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe merged pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe merged PR #13759:
URL: https://github.com/apache/kafka/pull/13759
--
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] cmccabe commented on a diff in pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1210806096
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -1106,6 +1107,34 @@ class KRaftClusterTest {
cluster.close()
}
}
+
+ @Test
+ def testTimedOutHeartbeats(): Unit = {
+ val cluster = new KafkaClusterTestKit.Builder(
+ new TestKitNodes.Builder().
+ setNumBrokerNodes(3).
+ setNumControllerNodes(1).build()).
+ setConfigProp(KafkaConfig.BrokerHeartbeatIntervalMsProp, 10.toString).
+ setConfigProp(KafkaConfig.BrokerSessionTimeoutMsProp, 1000.toString).
+ build()
+ try {
+ cluster.format()
+ cluster.startup()
+ val controller = cluster.controllers().values().iterator().next()
+ controller.controller.waitForReadyBrokers(3).get()
+ TestUtils.retry(60000) {
+ val latch = controller.controller.asInstanceOf[QuorumController].pause()
+ Thread.sleep(1001)
+ latch.countDown()
+ controller.controller.asInstanceOf[QuorumController].pause().countDown()
Review Comment:
I'll add a better explanation for this
--
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] ijuma commented on pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #13759:
URL: https://github.com/apache/kafka/pull/13759#issuecomment-1566213048
Thanks for the PR. Can you explain how this helps in a bit more detail? Let's use a small cluster with 3 or 4 brokers as an example. The 3 missed heartbeats in your example would result in 3 brokers being fenced by the time the overload state is entered? Or did I misunderstand? And what happens next?
--
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] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of broker heartbeat timeouts
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1210720723
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -1106,6 +1107,34 @@ class KRaftClusterTest {
cluster.close()
}
}
+
+ @Test
+ def testTimedOutHeartbeats(): Unit = {
+ val cluster = new KafkaClusterTestKit.Builder(
+ new TestKitNodes.Builder().
+ setNumBrokerNodes(3).
+ setNumControllerNodes(1).build()).
+ setConfigProp(KafkaConfig.BrokerHeartbeatIntervalMsProp, 10.toString).
+ setConfigProp(KafkaConfig.BrokerSessionTimeoutMsProp, 1000.toString).
+ build()
+ try {
+ cluster.format()
+ cluster.startup()
+ val controller = cluster.controllers().values().iterator().next()
+ controller.controller.waitForReadyBrokers(3).get()
+ TestUtils.retry(60000) {
+ val latch = controller.controller.asInstanceOf[QuorumController].pause()
+ Thread.sleep(1001)
+ latch.countDown()
+ controller.controller.asInstanceOf[QuorumController].pause().countDown()
Review Comment:
What is the purpose of this line? Seems it is an unnecessary extra pause/resume, and the test passes without it. Can we remove it?
##########
metadata/src/main/java/org/apache/kafka/controller/errors/ControllerExceptions.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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.controller.errors;
+
+import org.apache.kafka.common.errors.TimeoutException;
+
+import java.util.concurrent.ExecutionException;
+
+
+public class ControllerExceptions {
+ /**
+ * Check if an exception is a normal timeout exception.
+ *
+ * @param exception The exception to check.
+ * @return True if the exception is a normal timeout exception and NOT a timeout
+ * exception generated by the controller shutting down.
+ */
+ public static boolean isNormalTimeoutException(Throwable exception) {
+ if (exception == null) return false;
+ if (exception instanceof ExecutionException) {
+ exception = exception.getCause();
+ if (exception == null) return false;
+ }
+ if (!(exception instanceof TimeoutException)) return false;
+ if (exception.getMessage() != null &&
+ exception.getMessage().equals("The controller is shutting down.")) {
+ return false;
+ }
+ return true;
Review Comment:
This seems brittle. Is there a way we can identify the specific thing we are interested in?
--
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] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13759:
URL: https://github.com/apache/kafka/pull/13759#discussion_r1206028552
##########
metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java:
##########
@@ -223,6 +283,21 @@ public BrokerHeartbeatState next() {
}
}
+ /**
+ * The maximum number of timed out heartbeats to count.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_MAX = 1000;
+
+ /**
+ * The time period over which to track timed out heartbeats.
+ */
+ static final long DEFAULT_TIMED_OUT_HEARTBEAT_COUNT_WINDOW_NS = TimeUnit.MINUTES.toNanos(5);
+
+ /**
+ * The number of heartbeats to notice missing before we go into overload.
+ */
+ static final int DEFAULT_TIMED_OUT_HEARTBEAT_OVERLOAD_THRESHOLD = 3;
+
Review Comment:
Whatever we set this to (and currently we always take this default value) will determine how long it would take for a controller to leave the overloaded state, right? So if we see a series of missed heartbeats that puts us into the overloaded state, and then we don't see any more missed heartbeats, it would take this amount of time to leave that overloaded state?
Consider the following case. We are overloaded. And then some broker crashes. How long would we want it to take before we move leadership away for any partitions led by that suddenly-dead broker? Ideally the session timeout, of course, but that won't happen if we are overloaded -- will it take 5 minutes as currently coded? That seems like a long time.
I think the only thing we can say is that we should move leadership away as soon as we have a correct view of the cluster. The question then becomes: how do we know when we have good information?
A fixed amount of time like this with no missed heartbeats -- that seems too simple. We likely have a reasonable view of the cluster if we see enough heartbeats without missing any such that it is likely that all brokers have had a chance to communicate with us. So it seems like maybe the broker heartbeat interval should factor into it somehow?
Let's say the broker heartbeat interval is 2 seconds. Right now we consider a heartbeat missed if it is more than half the heartbeat interval (i.e. 1 second) old. Let's say there are 60 brokers in the cluster. If we see something on the order of 60 heartbeats in row without having missing any, then we probably have a reasonable view.
So maybe it isn't having missed no more than N heartbeats in some fixed time window. Maybe we define "not overloaded" as having seen some string of heartbeats uninterrupted without seeing a missed one? So every time we miss a heartbeat we go into the "overloaded" state and we reset a counter of contiguous successfully processed heartbeats to 0. Whenever we see a heartbeat and process it in time we increase that counter. We are no longer overloaded when the `contiguousHeartbeatsProcessedInTime` counter is equal to or exceeds the broker count.
I'm not sure I have it right, but I think this is a discussion worth pursuing.
--
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