Skip to content

KAFKA-8755: Fix state restore for standby tasks with optimized topology #7238

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 6 commits into from
Sep 13, 2019
Merged

KAFKA-8755: Fix state restore for standby tasks with optimized topology #7238

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
e468587
KAFKA-8755: Fix state restore for standby tasks with optimized topology
cpettitt-confluent Aug 27, 2019
3d92400
Don't restore unnecessarily on standby -> active task transition
cpettitt-confluent Sep 4, 2019
1f8a06b
Fixes for review feedback
cpettitt-confluent Sep 5, 2019
390606c
Fix NPath checkstyle error
cpettitt-confluent Sep 5, 2019
7a18866
More review fixes
cpettitt-confluent Sep 6, 2019
0b14fbd
More review feedback fixes
cpettitt-confluent Sep 11, 2019
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
37 changes: 14 additions & 23 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,26 +171,6 @@ public String toString(final String indent) {
return sb.toString();
}

protected void updateOffsetLimits() {
for (final TopicPartition partition : partitions) {
try {
final OffsetAndMetadata metadata = consumer.committed(partition); // TODO: batch API?
final long offset = metadata != null ? metadata.offset() : 0L;
stateMgr.putOffsetLimit(partition, offset);

if (log.isTraceEnabled()) {
log.trace("Updating store offset limits {} for changelog {}", offset, partition);
}
} catch (final AuthorizationException e) {
throw new ProcessorStateException(String.format("task [%s] AuthorizationException when initializing offsets for %s", id, partition), e);
} catch (final WakeupException e) {
throw e;
} catch (final KafkaException e) {
throw new ProcessorStateException(String.format("task [%s] Failed to initialize offsets for %s", id, partition), e);
}
}
}

/**
* Flush all state stores owned by this task
*/
Expand Down Expand Up @@ -219,9 +199,6 @@ void registerStateStores() {
}
log.trace("Initializing state stores");

// set initial offset limits
updateOffsetLimits();

for (final StateStore store : topology.stateStores()) {
log.trace("Initializing store {}", store.name());
processorContext.uninitialize();
Expand Down Expand Up @@ -272,4 +249,18 @@ public boolean hasStateStores() {
public Collection<TopicPartition> changelogPartitions() {
return stateMgr.changelogPartitions();
}

long committedOffsetForPartition(final TopicPartition partition) {
try {
final OffsetAndMetadata metadata = consumer.committed(partition);

@mjsax mjsax Sep 6, 2019 •
edited
Loading

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.

@guozhangwang With regard to you comment about exceptions for EOS -- it seem only relevant to catch and swallow a TimeoutException here? Or should the caller be responsible to decide if a timeout should be swallowed? Or do you have anything else in mind?

@guozhangwang guozhangwang Sep 6, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I think TimeoutException is the one, yes, but more importantly we should see if our timeout values are set reasonably in Streams :)

return metadata != null ? metadata.offset() : 0L;
} catch (final AuthorizationException e) {
throw new ProcessorStateException(String.format("task [%s] AuthorizationException when initializing offsets for %s", id, partition), e);
} catch (final WakeupException e) {
throw e;
} catch (final KafkaException e) {
throw new ProcessorStateException(String.format("task [%s] Failed to initialize offsets for %s", id, partition), e);
}
}

}
10 changes: 10 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStandbyTasks.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,4 +24,14 @@ class AssignedStandbyTasks extends AssignedTasks<StandbyTask> {
super(logContext, "standby task");
}

@Override
int commit() {
final int committed = super.commit();
// TODO: this contortion would not be necessary if we got rid of the two-step
// task.commitNeeded and task.commit and instead just had task.commitIfNeeded. Currently
// we only call commit if commitNeeded is true, which means that we need a way to indicate
// that we are eligible for updating the offset limit outside of commit.
running.forEach((id, task) -> task.allowUpdateOfOffsetLimit());

@mjsax mjsax Sep 6, 2019

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.

Seems we should add AssignedStandbyTasksTest.java to unit test the new commit() behavior.

@cpettitt-confluent cpettitt-confluent Sep 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

We could add a test here, but it is super implementation specific. This is not really the best home for the code, just the most convenient at the moment. All we would test, I think, is that we called allowUpdateOfOffsetLimit? The integration test covers this in a more comprehensive way and doesn't break when we move this code. What do you think?

@mjsax mjsax Sep 10, 2019

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.

All we would test, I think, is that we called allowUpdateOfOffsetLimit?

Yes, the test should be shouldUpdateOffsetsLimtOnCommit().

The integration test covers this in a more comprehensive way and doesn't break when we move this code.

"and doesn't break" -- I guess it should break? The main purpose of the unit test is, that is would be hard to debug the integration test and figure out the root cause why it fails.

@cpettitt-confluent cpettitt-confluent Sep 11, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I added back the StandbyTask tests that will cover this with a much more narrow test - specifically to cover when to update the offset limit. If this gets moved into StandbyTask that test should not fail, provided it was done right. If it's broken in any way that test will flag it and it should be easy to look at the test failure and the changed code to determine the cause.

Do you think StandbyTaskTest sufficiently addresses your concern?

return committed;
}
}
67 changes: 52 additions & 15 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 @@ -16,6 +16,14 @@
*/
package org.apache.kafka.streams.processor.internals;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
Expand All @@ -25,20 +33,14 @@
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
* A StandbyTask
*/
public class StandbyTask extends AbstractTask {
Comment thread
cpettitt-confluent marked this conversation as resolved.

private Map<TopicPartition, Long> checkpointedOffsets = new HashMap<>();
private final Sensor closeTaskSensor;
private final Map<TopicPartition, Long> offsetLimits = new HashMap<>();
private final Set<TopicPartition> updateableOffsetLimits = new HashSet<>();

/**
* Create {@link StandbyTask} with its assigned partitions
Expand All @@ -63,6 +65,14 @@ public class StandbyTask extends AbstractTask {

closeTaskSensor = metrics.threadLevelSensor("task-closed", Sensor.RecordingLevel.INFO);
processorContext = new StandbyContextImpl(id, config, stateMgr, metrics);

final Set<String> changelogTopicNames = new HashSet<>(topology.storeToChangelogTopic().values());
partitions.stream()
.filter(tp -> changelogTopicNames.contains(tp.topic()))
.forEach(tp -> {
offsetLimits.put(tp, 0L);
updateableOffsetLimits.add(tp);
});
}

@Override
Expand All @@ -88,7 +98,7 @@ public void initializeTopology() {
@Override
public void resume() {
log.debug("Resuming");
updateOffsetLimits();
allowUpdateOfOffsetLimit();
}

/**
Expand All @@ -102,9 +112,7 @@ public void resume() {
public void commit() {
log.trace("Committing");
flushAndCheckpointState();
// reinitialize offset limits
updateOffsetLimits();

allowUpdateOfOffsetLimit();
commitNeeded = false;
}

Expand Down Expand Up @@ -165,14 +173,25 @@ public void closeSuspended(final boolean clean,
*/
public List<ConsumerRecord<byte[], byte[]>> update(final TopicPartition partition,
final List<ConsumerRecord<byte[], byte[]>> records) {
if (records.isEmpty()) {
return Collections.emptyList();
}

log.trace("Updating standby replicas of its state store for partition [{}]", partition);
final long limit = stateMgr.offsetLimit(partition);
long limit = offsetLimits.getOrDefault(partition, Long.MAX_VALUE);
Comment thread
cpettitt-confluent marked this conversation as resolved.

long lastOffset = -1L;
final List<ConsumerRecord<byte[], byte[]>> restoreRecords = new ArrayList<>(records.size());
final List<ConsumerRecord<byte[], byte[]>> remainingRecords = new ArrayList<>();

for (final ConsumerRecord<byte[], byte[]> record : records) {
// Check if we're unable to process records due to an offset limit (e.g. when our
// partition is both a source and a changelog). If we're limited then try to refresh
// the offset limit if possible.
if (record.offset() >= limit && updateableOffsetLimits.contains(partition)) {
limit = updateOffsetLimits(partition);
Comment thread
cpettitt-confluent marked this conversation as resolved.
}

if (record.offset() < limit) {
restoreRecords.add(record);
lastOffset = record.offset();
Expand All @@ -181,9 +200,8 @@ public List<ConsumerRecord<byte[], byte[]>> update(final TopicPartition partitio
}
}

stateMgr.updateStandbyStates(partition, restoreRecords, lastOffset);

if (!restoreRecords.isEmpty()) {
stateMgr.updateStandbyStates(partition, restoreRecords, lastOffset);
Comment thread
cpettitt-confluent marked this conversation as resolved.
commitNeeded = true;
}

Expand All @@ -194,4 +212,23 @@ Map<TopicPartition, Long> checkpointedOffsets() {
return checkpointedOffsets;
}

private long updateOffsetLimits(final TopicPartition partition) {
if (!offsetLimits.containsKey(partition)) {
throw new IllegalArgumentException("Topic is not both a source and a changelog: " + partition);
}

updateableOffsetLimits.remove(partition);

final long newLimit = committedOffsetForPartition(partition);
final long previousLimit = offsetLimits.put(partition, newLimit);
if (previousLimit > newLimit) {
throw new IllegalStateException("Offset limit should monotonically increase, but was reduced. " +
"New limit: " + newLimit + ". Previous limit: " + previousLimit);
}
return newLimit;
}

void allowUpdateOfOffsetLimit() {
updateableOffsetLimits.addAll(offsetLimits.keySet());
}
}
60 changes: 33 additions & 27 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.processor.internals;

import java.util.Map.Entry;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
Expand Down Expand Up @@ -44,7 +45,7 @@ public class StoreChangelogReader implements ChangelogReader {
private final Logger log;
private final Consumer<byte[], byte[]> restoreConsumer;
private final StateRestoreListener userStateRestoreListener;
private final Map<TopicPartition, Long> endOffsets = new HashMap<>();
private final Map<TopicPartition, Long> restoreToOffsets = new HashMap<>();
private final Map<String, List<PartitionInfo>> partitionInfo = new HashMap<>();
private final Map<TopicPartition, StateRestorer> stateRestorers = new HashMap<>();
private final Set<TopicPartition> needsRestoring = new HashSet<>();
Expand Down Expand Up @@ -89,11 +90,11 @@ public Collection<TopicPartition> restore(final RestoringTasks active) {

for (final TopicPartition partition : needsRestoring) {
final StateRestorer restorer = stateRestorers.get(partition);
final long pos = processNext(records.records(partition), restorer, endOffsets.get(partition));
final long pos = processNext(records.records(partition), restorer, restoreToOffsets.get(partition));
restorer.setRestoredOffset(pos);
if (restorer.hasCompleted(pos, endOffsets.get(partition))) {
if (restorer.hasCompleted(pos, restoreToOffsets.get(partition))) {
restorer.restoreDone();
endOffsets.remove(partition);
restoreToOffsets.remove(partition);
completedRestorers.add(partition);
}
}
Expand Down Expand Up @@ -141,39 +142,44 @@ private void initialize(final RestoringTasks active) {

// try to fetch end offsets for the initializable restorers and remove any partitions
// where we already have all of the data
final Map<TopicPartition, Long> endOffsets;
try {
endOffsets.putAll(restoreConsumer.endOffsets(initializable));
endOffsets = restoreConsumer.endOffsets(initializable);
} catch (final TimeoutException e) {
// if timeout exception gets thrown we just give up this time and retry in the next run loop
log.debug("Could not fetch end offset for {}; will fall back to partition by partition fetching", initializable);
return;
}

endOffsets.forEach((partition, endOffset) -> {
if (endOffset != null) {
final StateRestorer restorer = stateRestorers.get(partition);
final long offsetLimit = restorer.offsetLimit();
restoreToOffsets.put(partition, Math.min(endOffset, offsetLimit));
} else {
log.info("End offset cannot be found form the returned metadata; removing this partition from the current run loop");
initializable.remove(partition);
}
});

final Iterator<TopicPartition> iter = initializable.iterator();
while (iter.hasNext()) {
final TopicPartition topicPartition = iter.next();
final Long endOffset = endOffsets.get(topicPartition);
final Long restoreOffset = restoreToOffsets.get(topicPartition);
final StateRestorer restorer = stateRestorers.get(topicPartition);

// offset should not be null; but since the consumer API does not guarantee it
// we add this check just in case
if (endOffset != null) {
final StateRestorer restorer = stateRestorers.get(topicPartition);
if (restorer.checkpoint() >= endOffset) {
restorer.setRestoredOffset(restorer.checkpoint());
iter.remove();
completedRestorers.add(topicPartition);
} else if (restorer.offsetLimit() == 0 || endOffset == 0) {
restorer.setRestoredOffset(0);
iter.remove();
completedRestorers.add(topicPartition);
} else {
restorer.setEndingOffset(endOffset);
}
needsInitializing.remove(topicPartition);
} else {
log.info("End offset cannot be found form the returned metadata; removing this partition from the current run loop");
if (restorer.checkpoint() >= restoreOffset) {
restorer.setRestoredOffset(restorer.checkpoint());
iter.remove();
completedRestorers.add(topicPartition);
} else if (restoreOffset == 0) {
restorer.setRestoredOffset(0);
iter.remove();
completedRestorers.add(topicPartition);
} else {
restorer.setEndingOffset(restoreOffset);
}
needsInitializing.remove(topicPartition);
}

// set up restorer for those initializable
Expand All @@ -200,7 +206,7 @@ private void startRestoration(final Set<TopicPartition> initialized,
restoreConsumer.seek(partition, restorer.checkpoint());
logRestoreOffsets(partition,
restorer.checkpoint(),
endOffsets.get(partition));
restoreToOffsets.get(partition));
restorer.setStartingOffset(restoreConsumer.position(partition));
restorer.restoreStarted();
} else {
Expand Down Expand Up @@ -232,7 +238,7 @@ private void startRestoration(final Set<TopicPartition> initialized,
final long position = restoreConsumer.position(restorer.partition());
logRestoreOffsets(restorer.partition(),
position,
endOffsets.get(restorer.partition()));
restoreToOffsets.get(restorer.partition()));
restorer.setStartingOffset(position);
restorer.restoreStarted();
}
Expand Down Expand Up @@ -279,7 +285,7 @@ public void reset() {
partitionInfo.clear();
stateRestorers.clear();
needsRestoring.clear();
endOffsets.clear();
restoreToOffsets.clear();
needsInitializing.clear();
completedRestorers.clear();
}
Expand Down
Loading