Skip to content

KAFKA-10185: Restoration info logging #8896

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
vvcephei merged 1 commit into from
Jun 19, 2020
Merged

KAFKA-10185: Restoration info logging #8896

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
6 changes: 6 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/errors/TaskCorruptedException.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.errors;

import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.streams.processor.TaskId;

Expand All @@ -36,7 +37,12 @@ public class TaskCorruptedException extends StreamsException {

public TaskCorruptedException(final Map<TaskId, Collection<TopicPartition>> taskWithChangelogs) {
super("Tasks with changelogs " + taskWithChangelogs + " are corrupted and hence needs to be re-initialized");
this.taskWithChangelogs = taskWithChangelogs;
}

public TaskCorruptedException(final Map<TaskId, Collection<TopicPartition>> taskWithChangelogs,
final InvalidOffsetException e) {
super("Tasks with changelogs " + taskWithChangelogs + " are corrupted and hence needs to be re-initialized", e);
this.taskWithChangelogs = taskWithChangelogs;
}

Expand Down
81 changes: 66 additions & 15 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 @@ -63,6 +63,8 @@
* be completed, while standby tasks updating changelog would always be in restoring state after being initialized.
*/
public class StoreChangelogReader implements ChangelogReader {
private static final long RESTORE_LOG_INTERVAL_MS = 10_000L;
private long lastRestoreLogTime = 0L;

enum ChangelogState {
// registered but need to be initialized (i.e. set its starting, end, limit offsets)
Expand Down Expand Up @@ -291,6 +293,7 @@ private boolean hasRestoredToEnd(final ChangelogMetadata metadata) {
public void enforceRestoreActive() {
if (state != ChangelogReaderState.ACTIVE_RESTORING) {
log.debug("Transiting to restore active tasks: {}", changelogs);
lastRestoreLogTime = 0L;

// pause all partitions that are for standby tasks from the restore consumer
pauseChangelogsFromRestoreConsumer(standbyRestoringChangelogs());
Expand Down Expand Up @@ -427,19 +430,20 @@ public void restore() {
// for restoring active and updating standby we may prefer different poll time
// in order to make sure we call the main consumer#poll in time.
// TODO: once we move ChangelogReader to a separate thread this may no longer be a concern
polledRecords = restoreConsumer.poll(state.equals(ChangelogReaderState.STANDBY_UPDATING) ? Duration.ZERO : pollTime);
polledRecords = restoreConsumer.poll(state == ChangelogReaderState.STANDBY_UPDATING ? Duration.ZERO : pollTime);

@vvcephei vvcephei Jun 18, 2020

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.

trivial cleanup

} catch (final InvalidOffsetException e) {
log.warn("Encountered {} fetching records from restore consumer for partitions {}, it is likely that " +
log.warn("Encountered " + e.getClass().getName() +
" fetching records from restore consumer for partitions " + e.partitions() + ", it is likely that " +
"the consumer's position has fallen out of the topic partition offset range because the topic was " +
"truncated or compacted on the broker, marking the corresponding tasks as corrupted and re-initializing" +
" it later.", e.getClass().getName(), e.partitions());
" it later.", e);
Comment on lines 434 to 439

@vvcephei vvcephei Jun 18, 2020

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.

Added the exception itself as the "cause" of the warning. The actual message of the IOE is actually pretty good at explaining the root cause.

@guozhangwang guozhangwang Jun 19, 2020

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.

The exception message may not always contain the partitions() list, maybe we should still print that as part of warn log?

@vvcephei vvcephei Jun 19, 2020

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.

It is still there, on L424.

@guozhangwang guozhangwang Jun 19, 2020

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.

Ah got it, I'm still think about it as the string template and was overlooking that. SG.


final Map<TaskId, Collection<TopicPartition>> taskWithCorruptedChangelogs = new HashMap<>();
for (final TopicPartition partition : e.partitions()) {
final TaskId taskId = changelogs.get(partition).stateManager.taskId();
taskWithCorruptedChangelogs.computeIfAbsent(taskId, k -> new HashSet<>()).add(partition);
}
throw new TaskCorruptedException(taskWithCorruptedChangelogs);
throw new TaskCorruptedException(taskWithCorruptedChangelogs, e);

@vvcephei vvcephei Jun 18, 2020

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.

Also added the cause to the thrown exception.

} catch (final KafkaException e) {
throw new StreamsException("Restore consumer get unexpected error polling records.", e);
}
Expand All @@ -448,7 +452,7 @@ public void restore() {
bufferChangelogRecords(restoringChangelogByPartition(partition), polledRecords.records(partition));
}

for (final TopicPartition partition: restoringChangelogs) {
for (final TopicPartition partition : restoringChangelogs) {
// even if some partition do not have any accumulated data, we still trigger
// restoring since some changelog may not need to restore any at all, and the
// restore to end check needs to be executed still.
Expand All @@ -458,9 +462,48 @@ public void restore() {
}

maybeUpdateLimitOffsetsForStandbyChangelogs();

maybeLogRestorationProgress();

@vvcephei vvcephei Jun 18, 2020

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 is the main change. Once every ten seconds, we will log the progress for each active restoring changelog.

}
}

private void maybeLogRestorationProgress() {
if (state == ChangelogReaderState.ACTIVE_RESTORING) {
if (time.milliseconds() - lastRestoreLogTime > RESTORE_LOG_INTERVAL_MS) {
final Set<TopicPartition> topicPartitions = activeRestoringChangelogs();
if (!topicPartitions.isEmpty()) {
final StringBuilder builder = new StringBuilder().append("Restoration in progress for ")
.append(topicPartitions.size())
.append(" partitions.");
for (final TopicPartition partition : topicPartitions) {

@guozhangwang guozhangwang Jun 19, 2020

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.

nit: should we have a newline for each partition? Otherwise that ling maybe too long.

@vvcephei vvcephei Jun 19, 2020

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 thought about it; while it does make the logs easier to read, it makes them harder to search (as in grep, since the line that would match the query doesn't contain all the information.

We do have other places where we concatenate every topic-partition on a single line, eg in the StreamsPartitionAssignor, so I think if long lines were a problem, people would already be complaining.

@guozhangwang guozhangwang Jun 19, 2020

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.

Actually I do have complaints about the StreamsPartitionAssignor log entries haha :)

Anyways, I think grep a valid reason. My rationale was that when searching for this entry, most people would use "Restoration in progress for" and then manually check if the particular interested partition in the following line, but I guess I'm just biased because I'm not a heavy grep user.

It is a quite nit comment and I won't feel strong about it. Your call.

final ChangelogMetadata changelogMetadata = restoringChangelogByPartition(partition);
builder.append(" {")
.append(partition)
.append(": ")
.append("position=")
.append(getPositionString(partition, changelogMetadata))
.append(", end=")
.append(changelogMetadata.restoreEndOffset)
.append(", totalRestored=")
.append(changelogMetadata.totalRestored)
.append("}");
}
log.info(builder.toString());
lastRestoreLogTime = time.milliseconds();
}
}
} else {
lastRestoreLogTime = 0L;
}
}

private static String getPositionString(final TopicPartition partition,
final ChangelogMetadata changelogMetadata) {
final ProcessorStateManager stateManager = changelogMetadata.stateManager;
final Long offsets = stateManager.changelogOffsets().get(partition);
return offsets == null ? "unknown" : String.valueOf(offsets);
}

private void maybeUpdateLimitOffsetsForStandbyChangelogs() {
// we only consider updating the limit offset for standbys if we are not restoring active tasks
if (state == ChangelogReaderState.STANDBY_UPDATING &&
Expand Down Expand Up @@ -496,8 +539,9 @@ private void bufferChangelogRecords(final ChangelogMetadata changelogMetadata, f
} else {
changelogMetadata.bufferedRecords.add(record);
final long offset = record.offset();
if (changelogMetadata.restoreEndOffset == null || offset < changelogMetadata.restoreEndOffset)
if (changelogMetadata.restoreEndOffset == null || offset < changelogMetadata.restoreEndOffset) {
changelogMetadata.bufferedLimitIndex = changelogMetadata.bufferedRecords.size();
}
Comment on lines 542 to 544

@vvcephei vvcephei Jun 18, 2020

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've rolled back a bunch of accidental formatting changes, but left the ones that are actually code style compliance issues (like using brackets around conditional bodies).

}
}
}
Expand Down Expand Up @@ -572,8 +616,9 @@ private Map<TopicPartition, Long> committedOffsetForChangelogs(final Set<TopicPa
}

private Map<TopicPartition, Long> endOffsetForChangelogs(final Set<TopicPartition> partitions) {
if (partitions.isEmpty())
if (partitions.isEmpty()) {
return Collections.emptyMap();
}

try {
final ListOffsetsResult result = adminClient.listOffsets(
Expand Down Expand Up @@ -617,8 +662,9 @@ private void updateLimitOffsetsForStandbyChangelogs(final Map<TopicPartition, Lo
}

private void initializeChangelogs(final Set<ChangelogMetadata> newPartitionsToRestore) {
if (newPartitionsToRestore.isEmpty())
if (newPartitionsToRestore.isEmpty()) {
return;
}

// for active changelogs, we need to find their end offset before transit to restoring
// if the changelog is on source topic, then its end offset should be the minimum of
Expand All @@ -631,45 +677,50 @@ private void initializeChangelogs(final Set<ChangelogMetadata> newPartitionsToRe
final TopicPartition partition = metadata.storeMetadata.changelogPartition();

// TODO K9113: when TaskType.GLOBAL is added we need to modify this
if (metadata.stateManager.taskType() == Task.TaskType.ACTIVE)
if (metadata.stateManager.taskType() == Task.TaskType.ACTIVE) {
newPartitionsToFindEndOffset.add(partition);
}

if (metadata.stateManager.changelogAsSource(partition))
if (metadata.stateManager.changelogAsSource(partition)) {
newPartitionsToFindCommittedOffset.add(partition);
}
}

// NOTE we assume that all requested partitions will be included in the returned map for both end/committed
// offsets, i.e., it would not return partial result and would timeout if some of the results cannot be found
final Map<TopicPartition, Long> endOffsets = endOffsetForChangelogs(newPartitionsToFindEndOffset);
final Map<TopicPartition, Long> committedOffsets = committedOffsetForChangelogs(newPartitionsToFindCommittedOffset);

for (final TopicPartition partition: newPartitionsToFindEndOffset) {
for (final TopicPartition partition : newPartitionsToFindEndOffset) {
final ChangelogMetadata changelogMetadata = changelogs.get(partition);
final Long endOffset = endOffsets.get(partition);
final Long committedOffset = newPartitionsToFindCommittedOffset.contains(partition) ?
committedOffsets.get(partition) : Long.valueOf(Long.MAX_VALUE);

if (endOffset != null && committedOffset != null) {
if (changelogMetadata.restoreEndOffset != null)
if (changelogMetadata.restoreEndOffset != null) {
throw new IllegalStateException("End offset for " + partition +
" should only be initialized once. Existing value: " + changelogMetadata.restoreEndOffset +
", new value: (" + endOffset + ", " + committedOffset + ")");
}

changelogMetadata.restoreEndOffset = Math.min(endOffset, committedOffset);

log.debug("End offset for changelog {} initialized as {}.", partition, changelogMetadata.restoreEndOffset);
} else {
if (!newPartitionsToRestore.remove(changelogMetadata))
if (!newPartitionsToRestore.remove(changelogMetadata)) {
throw new IllegalStateException("New changelogs to restore " + newPartitionsToRestore +
" does not contain the one looking for end offset: " + partition + ", this should not happen.");
}

log.info("End offset for changelog {} cannot be found; will retry in the next time.", partition);
}
}

// try initialize limit offsets for standby tasks for the first time
if (!committedOffsets.isEmpty())
if (!committedOffsets.isEmpty()) {
updateLimitOffsetsForStandbyChangelogs(committedOffsets);
}

// add new partitions to the restore consumer and transit them to restoring state
addChangelogsToRestoreConsumer(newPartitionsToRestore.stream().map(metadata -> metadata.storeMetadata.changelogPartition())
Expand Down Expand Up @@ -743,7 +794,7 @@ private void prepareChangelogs(final Set<ChangelogMetadata> newPartitionsToResto
// separate those who do not have the current offset loaded from checkpoint
final Set<TopicPartition> newPartitionsWithoutStartOffset = new HashSet<>();

for (final ChangelogMetadata changelogMetadata: newPartitionsToRestore) {
for (final ChangelogMetadata changelogMetadata : newPartitionsToRestore) {
final StateStoreMetadata storeMetadata = changelogMetadata.storeMetadata;
final TopicPartition partition = storeMetadata.changelogPartition();
final Long currentOffset = storeMetadata.offset();
Expand Down
19 changes: 17 additions & 2 deletions .../src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,11 +59,14 @@
import java.util.function.Function;
import java.util.stream.Collectors;

import static java.util.Collections.singletonMap;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.common.utils.Utils.mkSet;
import static org.apache.kafka.streams.processor.internals.Task.TaskType.ACTIVE;
import static org.apache.kafka.streams.processor.internals.Task.TaskType.STANDBY;
import static org.apache.kafka.streams.processor.internals.StoreChangelogReader.ChangelogReaderState.ACTIVE_RESTORING;
import static org.apache.kafka.streams.processor.internals.StoreChangelogReader.ChangelogReaderState.STANDBY_UPDATING;
import static org.apache.kafka.streams.processor.internals.Task.TaskType.ACTIVE;
import static org.apache.kafka.streams.processor.internals.Task.TaskType.STANDBY;
import static org.apache.kafka.test.MockStateRestoreListener.RESTORE_BATCH;
import static org.apache.kafka.test.MockStateRestoreListener.RESTORE_END;
import static org.apache.kafka.test.MockStateRestoreListener.RESTORE_START;
Expand Down Expand Up @@ -223,6 +226,7 @@ public void shouldInitializeChangelogAndCheckForCompletion() {
@Test
public void shouldPollWithRightTimeout() {
EasyMock.expect(storeMetadata.offset()).andReturn(null).andReturn(9L).anyTimes();
EasyMock.expect(stateManager.changelogOffsets()).andReturn(singletonMap(tp, 5L));

@vvcephei vvcephei Jun 18, 2020

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 is moderately obnoxious... The addition of logging these values means that these tests will get a NullPointerException unless we mock this call, but the mock is irrelevant to the test outcome.

@guozhangwang guozhangwang Jun 19, 2020

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.

This comment seems worth adding to the code :)

@vvcephei vvcephei Jun 19, 2020

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 didn't think to do this... This might be equivocation, but it seems like if I wrote that in a code comment, it may or may not be true in the future. Looking at the tests, there are already like a dozen cryptic, redundant mocks, so I'm not sure justifying this one really makes a material impact on this test's readability, which is already approaching zero.

Adding a comment like "this is just to prevent the logger from throwing an NPE" carries the risk that it can quickly become untrue in two ways:

  1. Maybe we remove or change the log so that it wouldn't need this mock; since it's a "nice" mock, we'll never know. In fact, I can't verify this call because the way the logger is configured only to print every ten seconds makes the NPE nondeterministic. Plus, it's not great to verify stuff that is beside the point of the test.
  2. Maybe we change the implementation so that it actually does exercise this mocked behavior, then the comment will become untrue, but we may not even notice.

Typically, having this many specific and complex mocks in a test indicates that we shouldn't be using easymock, but instead configure the component with "dummy" state manager, etc. If we re-wrote this test to use that strategy, then we wouldn't need to make explicit expectations like this.

Anyway, that's why I'm sort of inclined on just declaring bankruptcy on the comprehensibility of this test.

EasyMock.replay(stateManager, storeMetadata, store);

consumer.updateBeginningOffsets(Collections.singletonMap(tp, 5L));
Expand All @@ -249,6 +253,7 @@ public void shouldPollWithRightTimeout() {
@Test
public void shouldRestoreFromPositionAndCheckForCompletion() {
EasyMock.expect(storeMetadata.offset()).andReturn(5L).anyTimes();
EasyMock.expect(stateManager.changelogOffsets()).andReturn(singletonMap(tp, 5L));
EasyMock.replay(stateManager, storeMetadata, store);

adminClient.updateEndOffsets(Collections.singletonMap(tp, 10L));
Expand Down Expand Up @@ -315,6 +320,7 @@ public void shouldRestoreFromPositionAndCheckForCompletion() {
@Test
public void shouldRestoreFromBeginningAndCheckCompletion() {
EasyMock.expect(storeMetadata.offset()).andReturn(null).andReturn(9L).anyTimes();
EasyMock.expect(stateManager.changelogOffsets()).andReturn(singletonMap(tp, 5L));
EasyMock.replay(stateManager, storeMetadata, store);

consumer.updateBeginningOffsets(Collections.singletonMap(tp, 5L));
Expand Down Expand Up @@ -411,6 +417,7 @@ public void shouldCheckCompletionIfPositionLargerThanEndOffset() {
@Test
public void shouldRequestPositionAndHandleTimeoutException() {
EasyMock.expect(storeMetadata.offset()).andReturn(10L).anyTimes();
EasyMock.expect(activeStateManager.changelogOffsets()).andReturn(singletonMap(tp, 10L));
EasyMock.replay(activeStateManager, storeMetadata, store);

final AtomicBoolean clearException = new AtomicBoolean(false);
Expand Down Expand Up @@ -472,6 +479,7 @@ public long position(final TopicPartition partition) {
@Test
public void shouldRequestEndOffsetsAndHandleTimeoutException() {
EasyMock.expect(storeMetadata.offset()).andReturn(5L).anyTimes();
EasyMock.expect(activeStateManager.changelogOffsets()).andReturn(singletonMap(tp, 5L));
EasyMock.replay(activeStateManager, storeMetadata, store);

final AtomicBoolean functionCalled = new AtomicBoolean(false);
Expand Down Expand Up @@ -541,6 +549,7 @@ public ListOffsetsResult listOffsets(final Map<TopicPartition, OffsetSpec> topic
public void shouldRequestCommittedOffsetsAndHandleTimeoutException() {
EasyMock.expect(stateManager.changelogAsSource(tp)).andReturn(true).anyTimes();
EasyMock.expect(storeMetadata.offset()).andReturn(5L).anyTimes();
EasyMock.expect(stateManager.changelogOffsets()).andReturn(singletonMap(tp, 5L));
EasyMock.replay(stateManager, storeMetadata, store);

final AtomicBoolean functionCalled = new AtomicBoolean(false);
Expand Down Expand Up @@ -849,6 +858,11 @@ public void shouldRestoreMultipleChangelogs() {
EasyMock.expect(storeMetadataTwo.offset()).andReturn(0L).anyTimes();
EasyMock.expect(activeStateManager.storeMetadata(tp1)).andReturn(storeMetadataOne).anyTimes();
EasyMock.expect(activeStateManager.storeMetadata(tp2)).andReturn(storeMetadataTwo).anyTimes();
EasyMock.expect(activeStateManager.changelogOffsets()).andReturn(mkMap(
mkEntry(tp, 5L),
mkEntry(tp1, 5L),
mkEntry(tp2, 5L)
)).anyTimes();
EasyMock.replay(activeStateManager, storeMetadata, store, storeMetadataOne, storeMetadataTwo);

setupConsumer(10, tp);
Expand Down Expand Up @@ -889,6 +903,7 @@ public void shouldTransitState() {
EasyMock.expect(storeMetadataTwo.offset()).andReturn(5L).anyTimes();
EasyMock.expect(standbyStateManager.storeMetadata(tp1)).andReturn(storeMetadataOne).anyTimes();
EasyMock.expect(standbyStateManager.storeMetadata(tp2)).andReturn(storeMetadataTwo).anyTimes();
EasyMock.expect(activeStateManager.changelogOffsets()).andReturn(singletonMap(tp, 5L));
EasyMock.replay(activeStateManager, standbyStateManager, storeMetadata, store, storeMetadataOne, storeMetadataTwo);

adminClient.updateEndOffsets(Collections.singletonMap(tp, 10L));
Expand Down