guozhangwang commented on a change in pull request #8896:
URL: https://github.com/apache/kafka/pull/8896#discussion_r442931356
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
##########
@@ -458,9 +462,48 @@ public void restore() {
}
maybeUpdateLimitOffsetsForStandbyChangelogs();
+
+ maybeLogRestorationProgress();
+ }
+ }
+
+ 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) {
Review comment:
nit: should we have a newline for each partition? Otherwise that ling
maybe too long.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
##########
@@ -415,19 +418,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);
} 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);
Review comment:
The exception message may not always contain the `partitions()` list,
maybe we should still print that as part of warn log?
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java
##########
@@ -223,6 +227,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));
Review comment:
This comment seems worth adding to the code :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
