You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "aliehsaeedii (via GitHub)" <gi...@apache.org> on 2024/09/06 11:42:20 UTC
[PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
aliehsaeedii opened a new pull request, #17116:
URL: https://github.com/apache/kafka/pull/17116
This PR implements exponential backoff for state directory lock to increase the time between two consecutive attempts of acquiring the lock.
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1761091432
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
Yes, I think we are talking about the same thing. All logic would be inside `TaskManager`.
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1760738049
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
It could be easier to move all the back-off logic to `TaskManager`. Before attempting to initialize in `addTaskToStateUpdater`, check `canTryLock`. If we catch a `LockException`, create the backoffRecord. Just an idea which maybe would not require changing many methods.
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #17116:
URL: https://github.com/apache/kafka/pull/17116#issuecomment-2338932712
Test failures seem related, @aliehsaeedii could you have a look? https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-17116/7/testReport/
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1760880614
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
I also wanted to propose something similar what Lucas wrote. I think that is simpler. But why not have a private method `canTryInitializeIfNeeded()` in `TaskManager` and create the backoff record when the `LockException lockException` is thrown? The initialization is the actual call we want to backoff, right?
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1747024247
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -681,4 +706,19 @@ public int hashCode() {
}
}
+ public static class BackoffRecord {
+ private static final long INITIAL_INTERVAL = 1;
+ private static final int MULTIPLIER = 2;
+ private static final long MAX_INTERVAL = 1000;
Review Comment:
should `MAX_INTERVAL` be bigger?
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1756867091
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
@aliehsaeedii @lucasbru I do not understand this branch.
If the task is not allowed to lock the state directory, the register method will just return as if the state directory were locked. That means, the task thinks it is initialized and the task is added to the state updater without the lock on the state directory. Since the task thinks it successfully registered the state store, it will never try to register the state store and acquire the lock on the state directory again. Thus, the task will be an illegal state, i.e., in restoration without the lock on the state directory.
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1756867091
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
@aliehsaeedii @lucasbru I do not understand this branch.
If the task is not allowed to lock the state directory, the register method will just return as if the state directory were locked. That means, the task thinks it is initialized and the task is added to the state updater without the lock on the state directory. Since the task thinks it successfully registered the state store, it will never try to register the state store and acquire the lock on the state directory again. Thus, the task will be an illegal 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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1747024247
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -681,4 +706,19 @@ public int hashCode() {
}
}
+ public static class BackoffRecord {
+ private static final long INITIAL_INTERVAL = 1;
+ private static final int MULTIPLIER = 2;
+ private static final long MAX_INTERVAL = 1000;
Review Comment:
`MAX_INTERVAL` be bigger?
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #17116:
URL: https://github.com/apache/kafka/pull/17116
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1761118028
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
this [PR](https://github.com/apache/kafka/pull/17209) may solve the issue. Could you please review 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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "chiacyu (via GitHub)" <gi...@apache.org>.
chiacyu commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1749084349
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -681,4 +698,24 @@ public int hashCode() {
}
}
+ public static class BackoffRecord {
+ private long attempts;
+ private long lastAttemptMs;
+ private static final ExponentialBackoff EXPONENTIAL_BACKOFF = new ExponentialBackoff(1, 2, 10000, 0.5);
Review Comment:
should we change it to "exponentialBackoff" to keep it uniform as previous instances.
https://github.com/apache/kafka/blob/a9f3ea2dd34f8704f880bf58af1ff3a3f1070fbb/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java#L30C40-L30C58
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1747481383
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -353,9 +356,11 @@ synchronized boolean lock(final TaskId taskId) {
if (lockOwner.equals(Thread.currentThread())) {
log.trace("{} Found cached state dir lock for task {}", logPrefix(), taskId);
// we already own the lock
+ lockedTasksToBackoffRecord.remove(taskId);
Review Comment:
can we ever have a backoff record in this branch?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,7 +93,9 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.debug("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
trace level, otherwise we'd be flooded again for debug logs
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -681,4 +701,18 @@ public int hashCode() {
}
}
+ public static class BackoffRecord {
+ private long attempts;
+ private long lastAttemptMs;
+ private static final ExponentialBackoff EXPONENTIAL_BACKOFF = new ExponentialBackoff(1, 2, 1000, 0.5);
+
+
+ public BackoffRecord() {
+ this.attempts = 0;
Review Comment:
shoudln't this initialize lastAttemptMs as well?
also, you seem to be creating this record on the first attempt. Then attempts should be set to 1, no?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,7 +93,9 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
Review Comment:
If we enter this branch, the next line that will be printed is "acquired state directory lock". But this is not what we want right? Don't we have to leave this method somehow?
We should also test this behavior in `StateManagerUtilTest`.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -365,10 +370,25 @@ synchronized boolean lock(final TaskId taskId) {
lockedTasksToOwner.put(taskId, Thread.currentThread());
// make sure the task directory actually exists, and create it if not
getOrCreateDirectoryForTask(taskId);
+ lockedTasksToBackoffRecord.remove(taskId);
return true;
}
}
+ public boolean canTryLock(final TaskId taskId, final long nowMs) {
+ return !lockedTasksToBackoffRecord.containsKey(taskId) || lockedTasksToBackoffRecord.get(taskId).canAttempt(nowMs);
+ }
+
+ private void updateOrCreateBackoffRecord(final TaskId taskId, final long nowMs) {
+ if (lockedTasksToBackoffRecord.containsKey(taskId)) {
+ final BackoffRecord backoffRecord = lockedTasksToBackoffRecord.get(taskId);
+ backoffRecord.lastAttemptMs = nowMs;
Review Comment:
nit: maybe we could move the update of the fields also into a method of BackoffRecord, something like backoffRecord.recordAttempt(nowMs); ?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -681,4 +701,18 @@ public int hashCode() {
}
}
+ public static class BackoffRecord {
+ private long attempts;
+ private long lastAttemptMs;
+ private static final ExponentialBackoff EXPONENTIAL_BACKOFF = new ExponentialBackoff(1, 2, 1000, 0.5);
Review Comment:
Agreed, once per second still seems like a lot. We could start with 10 seconds, and see if we want to increase it. wdyt, @cadonna ?
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1747024247
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -681,4 +706,19 @@ public int hashCode() {
}
}
+ public static class BackoffRecord {
+ private static final long INITIAL_INTERVAL = 1;
+ private static final int MULTIPLIER = 2;
+ private static final long MAX_INTERVAL = 1000;
Review Comment:
should `MAX_INTERVAL` be bigger?
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1761118028
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
this [PR](https://github.com/apache/kafka/pull/17209) may solve the issue. Could you please review it? @lucasbru
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1760719974
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
@cadonna @lucasbru Could `StateManagerUtil.registerStateStores()` return a boolean instead of void? This can solve the issue, otherwise, if we throw an exception in the case of `stateDirectory.canTryLock() = false,` what would be the benefit of having backoff-retry for the lock? However this is not a simple small change. I assume we need to change many methods then.
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1757137235
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", id);
Review Comment:
Good catch. I missed this during review. This needs to be fixed.
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1747469091
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -681,4 +701,18 @@ public int hashCode() {
}
}
+ public static class BackoffRecord {
+ private long attempts;
+ private long lastAttemptMs;
+ private static final ExponentialBackoff EXPONENTIAL_BACKOFF = new ExponentialBackoff(1, 2, 1000, 0.5);
Review Comment:
`MaxInterval` is set to `1000`. Should it be increased?
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1747024247
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -681,4 +706,19 @@ public int hashCode() {
}
}
+ public static class BackoffRecord {
+ private static final long INITIAL_INTERVAL = 1;
+ private static final int MULTIPLIER = 2;
+ private static final long MAX_INTERVAL = 1000;
Review Comment:
`MAX_INTERVAL` can be increased.
--
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
Re: [PR] KAFKA-17109: implement exponential backoff for state directory lock [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1750207959
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -681,4 +698,24 @@ public int hashCode() {
}
}
+ public static class BackoffRecord {
+ private long attempts;
+ private long lastAttemptMs;
+ private static final ExponentialBackoff EXPONENTIAL_BACKOFF = new ExponentialBackoff(1, 2, 10000, 0.5);
Review Comment:
Naming static variables is different.
--
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