abbccdda commented on a change in pull request #9047:
URL: https://github.com/apache/kafka/pull/9047#discussion_r460165601
##########
File path: docs/streams/developer-guide/config-streams.html
##########
@@ -326,13 +321,18 @@ <h4><a class="toc-backref"
href="#id5">bootstrap.servers</a><a class="headerlink
<tr class="row-even"><td>state.cleanup.delay.ms</td>
<td>Low</td>
<td colspan="2">The amount of time in milliseconds to wait before
deleting state when a partition has migrated.</td>
- <td>600000 milliseconds</td>
+ <td>600000 milliseconds (10 minutes)</td>
</tr>
<tr class="row-odd"><td>state.dir</td>
<td>High</td>
<td colspan="2">Directory location for state stores.</td>
<td><code class="docutils literal"><span
class="pre">/tmp/kafka-streams</span></code></td>
</tr>
+ <tr class="row-odd"><td>task.timeout.ms</td>
+ <td>Medium</td>
Review comment:
nit: for a doc clean-up, it is helpful to include a screenshot of
updated paragraph. The changes starting at L331 should be suffice.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -318,6 +341,72 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
}
}
+ private void retryUntilSuccessOrThrowOnTaskTimeout(final Runnable runnable,
+ final String
errorMessage) {
+ long deadlineMs = NO_DEADLINE;
+
+ do {
+ try {
+ runnable.run();
+ return;
+ } catch (final TimeoutException retryableException) {
Review comment:
`retriableException` to be consistent with the defined exception type in
AK.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -275,31 +259,70 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
final RecordConverter recordConverter) {
for (final TopicPartition topicPartition : topicPartitions) {
globalConsumer.assign(Collections.singletonList(topicPartition));
+ long offset;
final Long checkpoint = checkpointFileCache.get(topicPartition);
if (checkpoint != null) {
globalConsumer.seek(topicPartition, checkpoint);
+ offset = checkpoint;
} else {
globalConsumer.seekToBeginning(Collections.singletonList(topicPartition));
+ final AtomicLong position = new AtomicLong();
Review comment:
For the above call, was curious why we couldn't seek for all the topic
partitions that are missing positions here, instead of doing one by one look-up?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/internals/QuietStreamsConfig.java
##########
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.streams.internals;
-
-import org.apache.kafka.streams.StreamsConfig;
-
-import java.util.Map;
-
-/**
- * A {@link StreamsConfig} that does not log its configuration on construction.
- *
- * This producer cleaner output for unit tests using the {@code test-utils},
- * since logging the config is not really valuable in this context.
- */
-public class QuietStreamsConfig extends StreamsConfig {
Review comment:
What's the purpose for this move?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -275,31 +259,70 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
final RecordConverter recordConverter) {
for (final TopicPartition topicPartition : topicPartitions) {
globalConsumer.assign(Collections.singletonList(topicPartition));
+ long offset;
final Long checkpoint = checkpointFileCache.get(topicPartition);
if (checkpoint != null) {
globalConsumer.seek(topicPartition, checkpoint);
+ offset = checkpoint;
} else {
globalConsumer.seekToBeginning(Collections.singletonList(topicPartition));
+ final AtomicLong position = new AtomicLong();
+ retryUntilSuccessOrThrowOnTaskTimeout(
+ () ->
position.set(globalConsumer.position(topicPartition)),
+ String.format(
+ "Failed to get position for partition %s. The broker
may be transiently unavailable at the moment.",
+ topicPartition
+ )
+ );
+ offset = position.get();
}
- long offset = globalConsumer.position(topicPartition);
final Long highWatermark = highWatermarks.get(topicPartition);
final RecordBatchingStateRestoreCallback stateRestoreAdapter =
StateRestoreCallbackAdapter.adapt(stateRestoreCallback);
stateRestoreListener.onRestoreStart(topicPartition, storeName,
offset, highWatermark);
long restoreCount = 0L;
+ long deadlineMs = NO_DEADLINE;
while (offset < highWatermark) {
try {
final ConsumerRecords<byte[], byte[]> records =
globalConsumer.poll(pollTime);
+ if (records.isEmpty()) {
+ if (taskTimeoutMs == 0L) {
+ deadlineMs = maybeUpdateDeadlineOrThrow(
+ deadlineMs,
+ requestTimeoutMs,
+ new StreamsException(String.format(
+ "Global task did not make progress to
restore state. Retrying is disabled. You can enable it by setting `%s` to a
value larger than zero.",
+ StreamsConfig.TASK_TIMEOUT_MS_CONFIG
+ ))
+ );
+ } else {
+ deadlineMs =
maybeUpdateDeadlineOrThrow(deadlineMs);
+ }
+
+ continue;
+ }
+ deadlineMs = NO_DEADLINE;
+
final List<ConsumerRecord<byte[], byte[]>> restoreRecords
= new ArrayList<>();
for (final ConsumerRecord<byte[], byte[]> record :
records.records(topicPartition)) {
if (record.key() != null) {
restoreRecords.add(recordConverter.convert(record));
}
}
- offset = globalConsumer.position(topicPartition);
+ try {
+ offset = globalConsumer.position(topicPartition);
+ } catch (final TimeoutException error) {
+ // the `globalConsumer.position()` call should never
block, because we know that we did
+ // a successful `position()` call above for the
requested partition and thus the consumer
+ // should have a valid local position that it can
return immediately
+
+ // hence, a `TimeoutException` indicates a bug and
thus we rethrow it as fatal `IllegalStateException`
+ throw new IllegalStateException(error);
Review comment:
We should add some explanation in the illegal state exception for why
such timeout is fatal
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -318,6 +341,72 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
}
}
+ private void retryUntilSuccessOrThrowOnTaskTimeout(final Runnable runnable,
+ final String
errorMessage) {
+ long deadlineMs = NO_DEADLINE;
+
+ do {
+ try {
+ runnable.run();
+ return;
+ } catch (final TimeoutException retryableException) {
+ if (taskTimeoutMs == 0L) {
+ throw new StreamsException(
+ String.format(
+ "Retrying is disabled. You can enable it by
setting `%s` to a value larger than zero.",
+ StreamsConfig.TASK_TIMEOUT_MS_CONFIG
+ ),
+ retryableException
+ );
+ }
+
+ deadlineMs = maybeUpdateDeadlineOrThrow(deadlineMs);
+
+ log.debug(errorMessage, retryableException);
Review comment:
Should we log warning here instead? At least this is a timeout.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -299,7 +318,17 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
restoreRecords.add(recordConverter.convert(record));
}
}
- offset = globalConsumer.position(topicPartition);
+ try {
+ offset = globalConsumer.position(topicPartition);
+ } catch (final TimeoutException error) {
+ // the `globalConsumer.position()` call should never
block, because we know that we did
+ // a successful `position()` call above for the
requested partition and thus the consumer
+ // should have a valid local position that it can
return immediately
Review comment:
If we do have a valid position from previous `position` call, do we
still need to update the position here again?
##########
File path: streams/src/test/java/org/apache/kafka/test/NoOpReadOnlyStore.java
##########
@@ -78,6 +78,7 @@ public void init(final ProcessorContext context, final
StateStore root) {
new File(context.stateDir() + File.separator + name).mkdir();
}
this.initialized = true;
+ context.register(root, (k, v) -> { });
Review comment:
Why do we need this?
----------------------------------------------------------------
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:
us...@infra.apache.org
