Skip to content

KAFKA-10097: Internalize checkpoint data #8820

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
mjsax merged 3 commits into from
Jun 7, 2020
Merged

KAFKA-10097: Internalize checkpoint data #8820

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5b25a3a
internalize checkpoint
Jun 4, 2020
6b7bcd1
minor change
Jun 6, 2020
bc9c9da
only checkpoint struct
Jun 6, 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
8 changes: 2 additions & 6 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

/**
Expand Down Expand Up @@ -153,12 +152,10 @@ public void postCommit() {
}

@Override
public Map<TopicPartition, Long> prepareCloseClean() {
public void prepareCloseClean() {
prepareClose(true);

log.info("Prepared clean close");

return Collections.emptyMap();
}

@Override
Expand Down Expand Up @@ -199,8 +196,7 @@ private void prepareClose(final boolean clean) {
}

@Override
public void closeClean(final Map<TopicPartition, Long> checkpoint) {
Objects.requireNonNull(checkpoint);
public void closeClean() {
close(true);

log.info("Closed clean");
Expand Down
54 changes: 30 additions & 24 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,8 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
private boolean commitNeeded = false;
private boolean commitRequested = false;

private Map<TopicPartition, Long> checkpoint = null;

public StreamTask(final TaskId id,
final Set<TopicPartition> partitions,
final ProcessorTopology topology,
Expand Down Expand Up @@ -465,17 +467,15 @@ private Map<TopicPartition, Long> extractPartitionTimes() {
}

@Override
public Map<TopicPartition, Long> prepareCloseClean() {
final Map<TopicPartition, Long> checkpoint = prepareClose(true);
public void prepareCloseClean() {
prepareClose(true);

log.info("Prepared clean close");

return checkpoint;
}

@Override
public void closeClean(final Map<TopicPartition, Long> checkpoint) {
close(true, checkpoint);
public void closeClean() {
close(true);

log.info("Closed clean");
}
Expand All @@ -489,7 +489,8 @@ public void prepareCloseDirty() {

@Override
public void closeDirty() {
close(false, null);

close(false);

log.info("Closed dirty");
}
Expand All @@ -505,11 +506,10 @@ public void update(final Set<TopicPartition> topicPartitions, final ProcessorTop

@Override
public void closeAndRecycleState() {
final Map<TopicPartition, Long> checkpoint = prepareClose(true);
prepareClose(true);

writeCheckpointIfNeed();

if (checkpoint != null) {
stateMgr.checkpoint(checkpoint);
}
switch (state()) {
case CREATED:
case RUNNING:
Expand Down Expand Up @@ -546,14 +546,14 @@ public void closeAndRecycleState() {
* otherwise, just close open resources
* @throws TaskMigratedException if the task producer got fenced (EOS)
*/
private Map<TopicPartition, Long> prepareClose(final boolean clean) {
final Map<TopicPartition, Long> checkpoint;
private void prepareClose(final boolean clean) {
// Reset any previously scheduled checkpoint.

@abbccdda abbccdda Jun 6, 2020

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we could just reset any checkpoint at the beginning, so that we only do checkpointing if this call thinks so. @mjsax

@mjsax mjsax Jun 6, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That should work.

checkpoint = null;

switch (state()) {
case CREATED:
// the task is created and not initialized, just re-write the checkpoint file
checkpoint = Collections.emptyMap();

scheduleCheckpoint(emptyMap());
break;

case RUNNING:
Expand All @@ -562,32 +562,38 @@ private Map<TopicPartition, Long> prepareClose(final boolean clean) {
if (clean) {
stateMgr.flush();
recordCollector.flush();
checkpoint = checkpointableOffsets();
scheduleCheckpoint(checkpointableOffsets());
} else {
checkpoint = null; // `null` indicates to not write a checkpoint

@mjsax mjsax Jun 6, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to set checkpointNeeded = false here?

executeAndMaybeSwallow(false, stateMgr::flush, "state manager flush", log);
}

break;

case RESTORING:
executeAndMaybeSwallow(clean, stateMgr::flush, "state manager flush", log);
checkpoint = Collections.emptyMap();
scheduleCheckpoint(emptyMap());

break;

case SUSPENDED:
case CLOSED:
// not need to checkpoint, since when suspending we've already committed the state
checkpoint = null; // `null` indicates to not write a checkpoint

@mjsax mjsax Jun 6, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As above?


break;

default:
throw new IllegalStateException("Unknown state " + state() + " while prepare closing active task " + id);
}
}

private void scheduleCheckpoint(final Map<TopicPartition, Long> checkpoint) {
this.checkpoint = checkpoint;
}

return checkpoint;
private void writeCheckpointIfNeed() {
if (checkpoint != null) {
stateMgr.checkpoint(checkpoint);
checkpoint = null;
}
}

/**
Expand All @@ -598,9 +604,9 @@ private Map<TopicPartition, Long> prepareClose(final boolean clean) {
* 3. finally release the state manager lock
* </pre>
*/
private void close(final boolean clean, final Map<TopicPartition, Long> checkpoint) {
if (clean && checkpoint != null) {
executeAndMaybeSwallow(clean, () -> stateMgr.checkpoint(checkpoint), "state manager checkpoint", log);
private void close(final boolean clean) {
if (clean) {
executeAndMaybeSwallow(true, this::writeCheckpointIfNeed, "state manager checkpoint", log);
}

switch (state()) {
Expand Down
4 changes: 2 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,12 +145,12 @@ enum TaskType {
*
* @throws StreamsException fatal error, should close the thread
*/
Map<TopicPartition, Long> prepareCloseClean();
void prepareCloseClean();

/**
* Must be idempotent.
*/
void closeClean(final Map<TopicPartition, Long> checkpoint);
void closeClean();

/**
* Prepare to close a task that we may not own. Discard any uncommitted progress and close the task.
Expand Down
33 changes: 14 additions & 19 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 @@ -189,7 +189,7 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
// first rectify all existing tasks
final LinkedHashMap<TaskId, RuntimeException> taskCloseExceptions = new LinkedHashMap<>();

final Map<Task, Map<TopicPartition, Long>> checkpointPerTask = new HashMap<>();
final Set<Task> tasksToClose = new HashSet<>();
final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();
final Set<Task> additionalTasksForCommitting = new HashSet<>();
final Set<Task> dirtyTasks = new HashSet<>();
Expand All @@ -209,11 +209,11 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
tasksToRecycle.add(task);
} else {
try {
final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
task.prepareCloseClean();
final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task
.committableOffsetsAndMetadata();

checkpointPerTask.put(task, checkpoint);
tasksToClose.add(task);
if (!committableOffsets.isEmpty()) {
consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
}
Expand Down Expand Up @@ -249,20 +249,17 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
log.error("Failed to batch commit tasks, " +
"will close all tasks involved in this commit as dirty by the end", e);
dirtyTasks.addAll(additionalTasksForCommitting);
dirtyTasks.addAll(checkpointPerTask.keySet());
dirtyTasks.addAll(tasksToClose);

checkpointPerTask.clear();
tasksToClose.clear();
// Just add first taskId to re-throw by the end.
taskCloseExceptions.put(consumedOffsetsAndMetadataPerTask.keySet().iterator().next(), e);
}
}

for (final Map.Entry<Task, Map<TopicPartition, Long>> taskAndCheckpoint : checkpointPerTask.entrySet()) {
final Task task = taskAndCheckpoint.getKey();
final Map<TopicPartition, Long> checkpoint = taskAndCheckpoint.getValue();

for (final Task task : tasksToClose) {
try {
completeTaskCloseClean(task, checkpoint);
completeTaskCloseClean(task);
cleanUpTaskProducer(task, taskCloseExceptions);
tasks.remove(task.id());
} catch (final RuntimeException e) {
Expand Down Expand Up @@ -630,9 +627,9 @@ private void closeTaskDirty(final Task task) {
task.closeDirty();
}

private void completeTaskCloseClean(final Task task, final Map<TopicPartition, Long> checkpoint) {
private void completeTaskCloseClean(final Task task) {
cleanupTask(task);
task.closeClean(checkpoint);
task.closeClean();
}

// Note: this MUST be called *before* actually closing the task
Expand All @@ -651,16 +648,16 @@ private void cleanupTask(final Task task) {
void shutdown(final boolean clean) {
final AtomicReference<RuntimeException> firstException = new AtomicReference<>(null);

final Map<Task, Map<TopicPartition, Long>> checkpointPerTask = new HashMap<>();
final Set<Task> tasksToClose = new HashSet<>();
final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();

for (final Task task : tasks.values()) {
if (clean) {
try {
final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
task.prepareCloseClean();
final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.committableOffsetsAndMetadata();

checkpointPerTask.put(task, checkpoint);
tasksToClose.add(task);
if (!committableOffsets.isEmpty()) {
consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
}
Expand All @@ -680,11 +677,9 @@ void shutdown(final boolean clean) {
commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
}

for (final Map.Entry<Task, Map<TopicPartition, Long>> taskAndCheckpoint : checkpointPerTask.entrySet()) {
final Task task = taskAndCheckpoint.getKey();
final Map<TopicPartition, Long> checkpoint = taskAndCheckpoint.getValue();
for (final Task task : tasksToClose) {
try {
completeTaskCloseClean(task, checkpoint);
completeTaskCloseClean(task);
} catch (final RuntimeException e) {
firstException.compareAndSet(null, e);
closeTaskDirty(task);
Expand Down
9 changes: 4 additions & 5 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@
import java.io.File;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;

import static java.util.Arrays.asList;
import static org.apache.kafka.common.utils.Utils.mkEntry;
Expand Down Expand Up @@ -270,8 +269,8 @@ public void shouldCommitOnCloseClean() {

task = createStandbyTask();
task.initializeIfNeeded();
final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
task.closeClean(checkpoint);
task.prepareCloseClean();
task.closeClean();

assertEquals(Task.State.CLOSED, task.state());

Expand Down Expand Up @@ -323,8 +322,8 @@ public void shouldThrowOnCloseCleanError() {
task = createStandbyTask();
task.initializeIfNeeded();

final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
assertThrows(RuntimeException.class, () -> task.closeClean(checkpoint));
task.prepareCloseClean();
assertThrows(RuntimeException.class, () -> task.closeClean());

final double expectedCloseTaskMetric = 0.0;
verifyCloseTaskMetric(expectedCloseTaskMetric, streamsMetrics, metricName);
Expand Down
26 changes: 13 additions & 13 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1567,8 +1567,8 @@ public void shouldCheckpointWithCreatedStateOnClose() {

task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);

final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
task.closeClean(checkpoint);
task.prepareCloseClean();
task.closeClean();

assertEquals(Task.State.CLOSED, task.state());
assertFalse(source1.initialized);
Expand Down Expand Up @@ -1642,8 +1642,8 @@ public void shouldNotCommitOnCloseRestoring() {
task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);

task.initializeIfNeeded();
final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
task.closeClean(checkpoint);
task.prepareCloseClean();
task.closeClean();

assertEquals(Task.State.CLOSED, task.state());

Expand All @@ -1668,8 +1668,8 @@ public void shouldCommitOnCloseClean() {
task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);
task.initializeIfNeeded();
task.completeRestoration();
final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
task.closeClean(checkpoint);
task.prepareCloseClean();
task.closeClean();

assertEquals(Task.State.CLOSED, task.state());

Expand All @@ -1696,8 +1696,8 @@ public void shouldSwallowExceptionOnCloseCleanError() {
task.initializeIfNeeded();
task.completeRestoration();

final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
assertThrows(ProcessorStateException.class, () -> task.closeClean(checkpoint));
task.prepareCloseClean();
assertThrows(ProcessorStateException.class, () -> task.closeClean());

final double expectedCloseTaskMetric = 0.0;
verifyCloseTaskMetric(expectedCloseTaskMetric, streamsMetrics, metricName);
Expand Down Expand Up @@ -1760,8 +1760,8 @@ public void shouldThrowOnCloseCleanCheckpointError() {
task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);
task.initializeIfNeeded();

final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
assertThrows(ProcessorStateException.class, () -> task.closeClean(checkpoint));
task.prepareCloseClean();
assertThrows(ProcessorStateException.class, () -> task.closeClean());

assertEquals(Task.State.RESTORING, task.state());

Expand Down Expand Up @@ -1795,11 +1795,11 @@ public void closeShouldBeIdempotent() {

task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);

final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
task.closeClean(checkpoint);
task.prepareCloseClean();
task.closeClean();

// close calls are idempotent since we are already in closed
task.closeClean(checkpoint);
task.closeClean();
task.closeDirty();

EasyMock.reset(stateManager);
Expand Down
Loading