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 2 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();
}

protected long getConsumerCommittedOffset(final TopicPartition partition) {
Comment thread
cpettitt-confluent marked this conversation as resolved.
Outdated
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;
}
}
64 changes: 49 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 @@ -166,12 +174,22 @@ public void closeSuspended(final boolean clean,
public List<ConsumerRecord<byte[], byte[]>> update(final TopicPartition partition,
final List<ConsumerRecord<byte[], byte[]>> records) {
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<>();

// Check if we are unable to process one or more records due to an offset limit (e.g. when
Comment thread
cpettitt-confluent marked this conversation as resolved.
Outdated
// our partition is both a source and changelog). If we are limited then try to refresh
// the offset limit if possible.
if (!records.isEmpty() &&

@guozhangwang guozhangwang Aug 26, 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 we only need to refresh offset limit if !remainingRecords.isEmpty meanings that some records cannot be updated, to replace the first two conditions.

Also since this should only happen when partition is part of sourceChangelogPartitions (otherwise offset limit should always be MAX_VALUE and the remaining records should always be empty), so instead of containing it as a condition, we can just check if it is not the case and throw indicating a bug.

@cpettitt-confluent cpettitt-confluent Aug 27, 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.

think we only need to refresh offset limit if !remainingRecords.isEmpty meanings that some records cannot be updated, to replace the first two conditions.

@guozhangwang Can do. This says something a little different than the current code. This says if there were any records that could not be played then refresh the consumer committed offset. The current code says refresh the consumer committed offset only if we could not play any of the records. In other words, we may refresh more frequently. Sound good?

Also since this should only happen when partition is part of sourceChangelogPartitions (otherwise offset limit should always be MAX_VALUE and the remaining records should always be empty), so instead of containing it as a condition, we can just check if it is not the case and throw indicating a bug.

Makes sense, I will treat this case as a bug and throw.

@cpettitt-confluent cpettitt-confluent Aug 28, 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 ended up not using remainingRecords.isEmpty, but am open to switching to that. Let me explain why I didn't pick this up. IIUC, the behavior you're looking for is to make the consumer.committed query if any records cannot be applied because the offset limit is too low. The new code does this. However, if I use remainingRecords.isEmpty I have to do up to 2 full passes on the records: once to determine that all records are blocked and the second to replay what is no longer blocked. Assuming records are ordered by offset we can achieve this in one pass as in the updated code. Please let me know my assumption is incorrect, I misunderstood your goal, or you'd still prefer the two pass approach.

@mjsax mjsax Sep 5, 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.

To simplify this, should we just check if records.isEmpty() in the very beginning and return immediately?

I also think, this code may still do two passed, as we have no guarantee that get() will not need to traverse the list anyway. To get a single pass, I would rather piggy pack updating the offset limit to the for loop below (we would change if to a proper Iterator that we advance manually: if we hit the else, we try to update the offset limit and retry the current record without advancing the iterator.

records.get(records.size() - 1).offset() >= limit &&
updateableOffsetLimits.remove(partition)) {
Comment thread
cpettitt-confluent marked this conversation as resolved.
Outdated

limit = updateOffsetLimits(partition);
}

for (final ConsumerRecord<byte[], byte[]> record : records) {
if (record.offset() < limit) {
restoreRecords.add(record);
Expand All @@ -181,9 +199,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 +211,21 @@ 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);
}

final long newLimit = getConsumerCommittedOffset(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());
}
}
43 changes: 24 additions & 19 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 @@ -141,39 +141,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> queriedEndOffsets;
try {
endOffsets.putAll(restoreConsumer.endOffsets(initializable));
queriedEndOffsets = 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;
}

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

final Iterator<TopicPartition> iter = initializable.iterator();
while (iter.hasNext()) {
final TopicPartition topicPartition = iter.next();
final Long endOffset = endOffsets.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");
final StateRestorer restorer = stateRestorers.get(topicPartition);
if (restorer.checkpoint() >= endOffset) {

@mjsax mjsax Sep 5, 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.

Thinking about this, the checkpoint should never be larger than the endOffset? Ie, is restorer.checkpoint() > endOffset actually an error condition and we should wipe out the state and recreate from scratch?

@cpettitt-confluent cpettitt-confluent Sep 5, 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.

Totally agree from a correctness perspective. Do you think we have enough test coverage to ensure this change is safe?

@cpettitt-confluent cpettitt-confluent Sep 5, 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.

Looking at this one a bit more, I think it might be a good idea to defer this change. It's pretty orthogonal to the changes I'm doing (this code was just moved in this patch). I can create a ticket if you like?

@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.

Maybe. Would be good to get input from @guozhangwang about it -- maybe >= is actually correct and I am just missing something...

restorer.setRestoredOffset(restorer.checkpoint());
iter.remove();
completedRestorers.add(topicPartition);
} else if (restorer.offsetLimit() == 0 || endOffset == 0) {

@mjsax mjsax Sep 5, 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.

This conditions seems to be redundant as we compute Math.min() above anyway. I think we can simplify to endOffset == 0.

Actually, I am wondering if we can unify the first and second branch:

final int currentOffset = Math.max(restorer.checkpoint(), 0);
if (currentOffset == endOffset) {
  restorer.setRestoredOffset(currentOffset);
  iter.remove();
 completedRestorers.add(topicPartition);
} else {
  restorer.setEndingOffset(endOffset);
}

@cpettitt-confluent cpettitt-confluent Sep 5, 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.

Sounds great!

restorer.setRestoredOffset(0);
iter.remove();
completedRestorers.add(topicPartition);
} else {
restorer.setEndingOffset(endOffset);
}
needsInitializing.remove(topicPartition);
}

// set up restorer for those initializable
Expand Down
42 changes: 29 additions & 13 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 @@ -16,6 +16,19 @@
*/
package org.apache.kafka.streams.processor.internals;

import static java.lang.String.format;
import static java.util.Collections.singleton;
import static org.apache.kafka.streams.kstream.internals.metrics.Sensors.recordLatenessSensor;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
Expand Down Expand Up @@ -48,18 +61,6 @@
import org.apache.kafka.streams.processor.internals.metrics.ThreadMetrics;
import org.apache.kafka.streams.state.internals.ThreadCache;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static java.lang.String.format;
import static java.util.Collections.singleton;
import static org.apache.kafka.streams.kstream.internals.metrics.Sensors.recordLatenessSensor;

/**
* A StreamTask is associated with a {@link PartitionGroup}, and is assigned to a StreamThread for processing.
*/
Expand Down Expand Up @@ -236,6 +237,22 @@ public StreamTask(final TaskId id,
@Override
public boolean initializeStateStores() {
log.trace("Initializing state stores");

// Currently there is no easy way to tell the ProcessorStateManager to only restore up to
// a specific offset. In most cases this is fine. However, in optimized topologies we can
// have a source topic that also serves as a changelog, and in this case we want our active
// stream task to only play records up to the last consumer committed offset. Here we find
// partitions of topics that are both sources and changelogs and set the consumer committed
// offset via stateMgr as there is not a more direct route.
final Set<String> changelogTopicNames = new HashSet<>(topology.storeToChangelogTopic().values());

@guozhangwang guozhangwang Sep 5, 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.

With EOS, Consumer.committed maybe taking long time as the previous txn is being completed, we need to make sure that it would not throw any unexpected exceptions and if so we should not fail the streams instance. cc @abbccdda @mjsax .

@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.

This code should be functionally equivalent to the previous code. Previously we initialized the offset limits in registerStateStores, which is called by initializeStateStores here. The biggest difference is that we no longer do this immediately for standby tasks - we defer to the first time we need a new offset limit to apply a record.

@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.

The only exception that is non-fatal should be TimeoutException?

@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.

Yeah this is not a comment for this PR but more or less an FYI for folks who are working on KIP-447 :) I also left some thoughts on the voting thread regarding this purpose.

partitions.stream()
.filter(tp -> changelogTopicNames.contains(tp.topic()))
.forEach(tp -> {
final long offset = getConsumerCommittedOffset(tp);
stateMgr.putOffsetLimit(tp, offset);
log.trace("Updating store offset limits {} for changelog {}", offset, tp);
});

registerStateStores();

return changelogPartitions().isEmpty();
Expand Down Expand Up @@ -460,7 +477,6 @@ void commit(final boolean startNewTransaction) {
final TopicPartition partition = entry.getKey();
final long offset = entry.getValue() + 1;
consumedOffsetsAndMetadata.put(partition, new OffsetAndMetadata(offset));
stateMgr.putOffsetLimit(partition, offset);
}

try {
Expand Down
Loading