You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "rondagostino (via GitHub)" <gi...@apache.org> on 2023/05/25 21:50:18 UTC
[GitHub] [kafka] rondagostino commented on a diff in pull request #13759: KAFKA-15019: Improve handling of overload situations in the kcontroller
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