Skip to content

KAFKA-10011: Remove task id from lockedTaskDirectories during handleLostAll #8682

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guozhangwang merged 9 commits into from
May 19, 2020
Merged

KAFKA-10011: Remove task id from lockedTaskDirectories during handleLostAll #8682

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
d2f2667
Handle task migrated inside corruption path
May 14, 2020
50bedd0
Merge remote-tracking branch 'upstream/trunk' into trunk
May 15, 2020
2c22da3
Merge remote-tracking branch 'upstream/trunk' into trunk
May 17, 2020
d9b38a6
locked task directory
May 17, 2020
48a2594
address comments
May 19, 2020
a27de55
log warn
guozhangwang May 19, 2020
a90e134
typo
guozhangwang May 19, 2020
01a5c55
reword a bit, change warn to info
guozhangwang May 19, 2020
3297542
revert
guozhangwang May 19, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -464,6 +464,7 @@ void handleLostAll() {
if (task.isActive()) {
closeTaskDirty(task);
iterator.remove();

try {
activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(task.id());
} catch (final RuntimeException e) {
Expand Down Expand Up @@ -516,6 +517,10 @@ public Map<TaskId, Long> getTaskOffsetSums() {
* assigned the task as a result of the rebalance). This method should be idempotent.
*/
private void tryToLockAllNonEmptyTaskDirectories() {
// Always clear the set at the beginning as we're always dealing with the
// current set of actually-locked tasks.
lockedTaskDirectories.clear();

for (final File dir : stateDirectory.listNonEmptyTaskDirectories()) {
try {
final TaskId id = TaskId.parse(dir.getName());
Expand Down
20 changes: 19 additions & 1 deletion streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -446,7 +446,7 @@ public Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
}

@Test
public void shouldCloseActiveTasksWhenHandlingLostTasks() {
public void shouldCloseActiveTasksWhenHandlingLostTasks() throws Exception {
final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
final Task task01 = new StateMachineTask(taskId01, taskId01Partitions, false);

Expand All @@ -457,6 +457,17 @@ public void shouldCloseActiveTasksWhenHandlingLostTasks() {
topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString());
expectLastCall().anyTimes();

makeTaskFolders(taskId00.toString(), taskId01.toString());
expectLockObtainedFor(taskId00, taskId01);

// The second attempt will return empty tasks.
makeTaskFolders();
expectLockObtainedFor();
replay(stateDirectory);

taskManager.handleRebalanceStart(emptySet());
assertThat(taskManager.lockedTaskDirectories(), Matchers.is(mkSet(taskId00, taskId01)));

// `handleLostAll`
activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(taskId00);
expectLastCall();
Expand All @@ -473,6 +484,13 @@ public void shouldCloseActiveTasksWhenHandlingLostTasks() {
assertThat(task01.state(), is(Task.State.RUNNING));
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), is(singletonMap(taskId01, task01)));

// The locked task map will not be cleared.
assertThat(taskManager.lockedTaskDirectories(), is(mkSet(taskId00, taskId01)));

taskManager.handleRebalanceStart(emptySet());

assertThat(taskManager.lockedTaskDirectories(), is(emptySet()));
}

@Test
Expand Down