You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2025/01/27 07:27:32 UTC
(kafka) branch 3.8 updated: KAFKA-17109: Implement exponential backoff for state directory lock
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.8 by this push:
new 84093c0b4f2 KAFKA-17109: Implement exponential backoff for state directory lock
84093c0b4f2 is described below
commit 84093c0b4f24bbfcf8c847ed0e0a205cb77077f1
Author: Alieh Saeedi <10...@users.noreply.github.com>
AuthorDate: Mon Jan 27 08:27:24 2025 +0100
KAFKA-17109: Implement exponential backoff for state directory lock
Cherry-pick of bb11257 and d04f534.
This implements exponential backoff for state directory lock to increase the time between two consecutive attempts of acquiring the lock.
Reviewers: Bruno Cadonna <ca...@apache.org>
---
.../processor/internals/StateDirectory.java | 3 --
.../streams/processor/internals/TaskManager.java | 49 ++++++++++++++++++++--
.../processor/internals/StreamThreadTest.java | 2 +-
.../processor/internals/TaskManagerTest.java | 48 +++++++++++++++++++++
4 files changed, 95 insertions(+), 7 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
index e76a94ef9f4..fe629eccfcf 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
@@ -362,8 +362,6 @@ public class StateDirectory implements AutoCloseable {
throw new IllegalStateException("The state directory has been deleted");
} else {
lockedTasksToOwner.put(taskId, Thread.currentThread());
- // make sure the task directory actually exists, and create it if not
- getOrCreateDirectoryForTask(taskId);
return true;
}
}
@@ -679,5 +677,4 @@ public class StateDirectory implements AutoCloseable {
return Objects.hash(file, namedTopology);
}
}
-
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
index fa0328447fd..561430bfea7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
@@ -31,6 +31,7 @@ import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.ExponentialBackoff;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.errors.LockException;
@@ -103,6 +104,8 @@ public class TaskManager {
// includes assigned & initialized tasks and unassigned tasks we locked temporarily during rebalance
private final Set<TaskId> lockedTaskDirectories = new HashSet<>();
+ private Map<TaskId, BackoffRecord> taskIdToBackoffRecord = new HashMap<>();
+
private final ActiveTaskCreator activeTaskCreator;
private final StandbyTaskCreator standbyTaskCreator;
private final StateUpdater stateUpdater;
@@ -1006,14 +1009,22 @@ public class TaskManager {
}
private void addTaskToStateUpdater(final Task task) {
+ final long nowMs = time.milliseconds();
try {
- task.initializeIfNeeded();
- stateUpdater.add(task);
+ if (canTryInitializeTask(task.id(), nowMs)) {
+ task.initializeIfNeeded();
+ taskIdToBackoffRecord.remove(task.id());
+ stateUpdater.add(task);
+ } else {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", task.id());
+ tasks.addPendingTasksToInit(Collections.singleton(task));
+ }
} catch (final LockException lockException) {
// The state directory may still be locked by another thread, when the rebalance just happened.
// Retry in the next iteration.
log.info("Encountered lock exception. Reattempting locking the state in the next iteration.", lockException);
tasks.addPendingTasksToInit(Collections.singleton(task));
+ updateOrCreateBackoffRecord(task.id(), nowMs);
}
}
@@ -1769,7 +1780,6 @@ public class TaskManager {
return standbyTasksInTaskRegistry;
}
}
-
// For testing only.
int commitAll() {
return commit(tasks.allTasks());
@@ -2116,4 +2126,37 @@ public class TaskManager {
void addTask(final Task task) {
tasks.addTask(task);
}
+
+ private boolean canTryInitializeTask(final TaskId taskId, final long nowMs) {
+ return !taskIdToBackoffRecord.containsKey(taskId) || taskIdToBackoffRecord.get(taskId).canAttempt(nowMs);
+ }
+
+ private void updateOrCreateBackoffRecord(final TaskId taskId, final long nowMs) {
+ if (taskIdToBackoffRecord.containsKey(taskId)) {
+ taskIdToBackoffRecord.get(taskId).recordAttempt(nowMs);
+ } else {
+ taskIdToBackoffRecord.put(taskId, new BackoffRecord(nowMs));
+ }
+ }
+
+ public static class BackoffRecord {
+ private long attempts;
+ private long lastAttemptMs;
+ private static final ExponentialBackoff EXPONENTIAL_BACKOFF = new ExponentialBackoff(1000, 2, 10000, 0.5);
+
+
+ public BackoffRecord(final long nowMs) {
+ this.attempts = 1;
+ this.lastAttemptMs = nowMs;
+ }
+
+ public void recordAttempt(final long nowMs) {
+ this.attempts++;
+ this.lastAttemptMs = nowMs;
+ }
+
+ public boolean canAttempt(final long nowMs) {
+ return nowMs - lastAttemptMs >= EXPONENTIAL_BACKOFF.backoff(attempts);
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index f0504859672..110a0e5e705 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -1134,7 +1134,7 @@ public class StreamThreadTest {
topologyMetadata.buildAndRewriteTopology();
final TaskManager taskManager = new TaskManager(
- null,
+ new MockTime(),
changelogReader,
null,
null,
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
index ee8d8f1da72..1c9b8d34e8d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
@@ -1250,6 +1250,54 @@ public class TaskManagerTest {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWithBackoffWhenInitializationFails() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ // task00 should not be initialized due to LockException, task01 should be initialized
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ time.sleep(500);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ // task00 should not be initialized since the backoff period has not passed
+ verify(task00, times(1)).initializeIfNeeded();
+ verify(tasks, times(2)).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00))
+ );
+ verify(stateUpdater, never()).add(task00);
+
+ time.sleep(5000);
+
+ // task00 should call initialize since the backoff period has passed
+ doNothing().when(task00).initializeIfNeeded();
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00, times(2)).initializeIfNeeded();
+ verify(tasks, times(2)).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00))
+ );
+ verify(stateUpdater).add(task00);
+ }
+
@Test
public void shouldRethrowRuntimeExceptionInInitTaskWithStateUpdater() {
final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)