mjsax commented on a change in pull request #10894:
URL: https://github.com/apache/kafka/pull/10894#discussion_r654155821
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -288,6 +288,19 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
// -> don't pass in `task.timeout.ms` to stay responsive if
`KafkaStreams#close` gets called
final ConsumerRecords<byte[], byte[]> records =
globalConsumer.poll(requestTimeoutPlusTaskTimeout);
if (records.isEmpty()) {
+ // get consumer position to step over potential TX markers
+ offset = retryUntilSuccessOrThrowOnTaskTimeout(
+ () -> globalConsumer.position(topicPartition),
+ String.format(
+ "Failed to get position for partition %s. The
broker may be transiently unavailable at the moment.",
+ topicPartition
+ )
+ );
+
+ if (offset == highWatermark) {
Review comment:
Could it mask a bug? `offset` should never be larger than
`highWatermark` ?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -288,6 +288,19 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
// -> don't pass in `task.timeout.ms` to stay responsive if
`KafkaStreams#close` gets called
final ConsumerRecords<byte[], byte[]> records =
globalConsumer.poll(requestTimeoutPlusTaskTimeout);
if (records.isEmpty()) {
+ // get consumer position to step over potential TX markers
+ offset = retryUntilSuccessOrThrowOnTaskTimeout(
+ () -> globalConsumer.position(topicPartition),
+ String.format(
+ "Failed to get position for partition %s. The
broker may be transiently unavailable at the moment.",
+ topicPartition
+ )
+ );
+
+ if (offset == highWatermark) {
Review comment:
Could it mask a bug? `offset` should never be larger than
`highWatermark` ?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -288,6 +288,19 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
// -> don't pass in `task.timeout.ms` to stay responsive if
`KafkaStreams#close` gets called
final ConsumerRecords<byte[], byte[]> records =
globalConsumer.poll(requestTimeoutPlusTaskTimeout);
if (records.isEmpty()) {
+ // get consumer position to step over potential TX markers
+ offset = retryUntilSuccessOrThrowOnTaskTimeout(
+ () -> globalConsumer.position(topicPartition),
+ String.format(
+ "Failed to get position for partition %s. The
broker may be transiently unavailable at the moment.",
+ topicPartition
+ )
+ );
+
+ if (offset == highWatermark) {
Review comment:
Good catch. I also updated the tests to cover this case.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/GlobalKTableEOSIntegrationTest.java
##########
@@ -339,17 +446,17 @@ public void shouldNotRestoreAbortedMessages() throws
Exception {
.getStore(globalStore, kafkaStreams,
QueryableStoreTypes.keyValueStore());
assertNotNull(store);
- final Map<Long, String> result = new HashMap<>();
+ final Map<Long, String> storeContent = new HashMap<>();
TestUtils.waitForCondition(
() -> {
- result.clear();
- store.all().forEachRemaining(pair -> result.put(pair.key,
pair.value));
- return result.equals(expected);
+ storeContent.clear();
+ store.all().forEachRemaining(pair ->
storeContent.put(pair.key, pair.value));
+ return storeContent.equals(expected);
},
- 30000L,
+ 30_000L,
() -> "waiting for initial values" +
"\n expected: " + expected +
- "\n received: " + results
Review comment:
side fix: this should have be `result` (singular) not `results` plural
to refer to the local variable instead of the global one. Renaming to avoid
this issue in the future.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -288,6 +288,22 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
// -> don't pass in `task.timeout.ms` to stay responsive if
`KafkaStreams#close` gets called
final ConsumerRecords<byte[], byte[]> records =
globalConsumer.poll(requestTimeoutPlusTaskTimeout);
Review comment:
Ah. Good point. We would need to refactor this anyway (cf the long
comment above). Seems it's time to do it now :)
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -288,6 +288,22 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
// -> don't pass in `task.timeout.ms` to stay responsive if
`KafkaStreams#close` gets called
final ConsumerRecords<byte[], byte[]> records =
globalConsumer.poll(requestTimeoutPlusTaskTimeout);
if (records.isEmpty()) {
+ // get consumer position to step over potential TX markers
+ offset = retryUntilSuccessOrThrowOnTaskTimeout(
+ () -> globalConsumer.position(topicPartition),
+ String.format(
Review comment:
Good idea!
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
##########
@@ -262,34 +260,19 @@ private void restoreState(final StateRestoreCallback
stateRestoreCallback,
stateRestoreListener.onRestoreStart(topicPartition, storeName,
offset, highWatermark);
long restoreCount = 0L;
- while (offset < highWatermark) { // when we "fix" this loop
(KAFKA-7380 / KAFKA-10317)
- // we should update the `poll()`
timeout below
-
- // we ignore `poll.ms` config during bootstrapping phase and
- // apply `request.timeout.ms` plus `task.timeout.ms` instead
- //
- // the reason is, that `poll.ms` might be too short to give a
fetch request a fair chance
- // to actually complete and we don't want to start
`task.timeout.ms` too early
- //
- // we also pass `task.timeout.ms` into `poll()` directly right
now as it simplifies our own code:
- // if we don't pass it in, we would just track the timeout
ourselves and call `poll()` again
- // in our own retry loop; by passing the timeout we can reuse
the consumer's internal retry loop instead
- //
- // note that using `request.timeout.ms` provides a
conservative upper bound for the timeout;
- // this implies that we might start `task.timeout.ms`
"delayed" -- however, starting the timeout
- // delayed is preferable (as it's more robust) than starting
it too early
+ while (offset < highWatermark) {
+ // we add `request.timeout.ms` to `poll.ms` because `poll.ms`
might be too short
+ // to give a fetch request a fair chance to actually complete
and we don't want to
+ // start `task.timeout.ms` too early
//
- // TODO https://issues.apache.org/jira/browse/KAFKA-10315
- // -> do a more precise timeout handling if `poll` would
throw an exception if a fetch request fails
- // (instead of letting the consumer retry fetch requests
silently)
- //
- // TODO https://issues.apache.org/jira/browse/KAFKA-10317 and
- // https://issues.apache.org/jira/browse/KAFKA-7380
- // -> don't pass in `task.timeout.ms` to stay responsive if
`KafkaStreams#close` gets called
- final ConsumerRecords<byte[], byte[]> records =
globalConsumer.poll(requestTimeoutPlusTaskTimeout);
+ // TODO with https://issues.apache.org/jira/browse/KAFKA-10315
we can just call
+ // `poll(pollMS)` without adding the request timeout and
do a more precise
+ // timeout handling
+ final ConsumerRecords<byte[], byte[]> records =
globalConsumer.poll(pollMsPlusRequestTimeout);
Review comment:
I still think we should pass `poll.ms` plus `request.timeout.ms` as long
as we don't have https://issues.apache.org/jira/browse/KAFKA-10315
This might still result in 30sec startup delay if `poll()` blocks as there
is no data, but I think it's acceptable (at least compared to 5 minutes...)
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -1044,7 +1043,7 @@ public synchronized long position(final TopicPartition
partition) {
}
@Test
- public void
shouldUseRequestTimeoutPlusTaskTimeoutInPollDuringRestoreAndFailIfNoDataReturned()
{
+ public void
shouldUsePollMsPlusRequestTimeoutInPollDuringRestoreAndTimeoutWhenNoProgressDuringRestore()
{
Review comment:
After the refactoring who we handle `task.timeout.ms` during global
restore, this test and the next test did cover the exact some thing. So I
removed one test to avoid redundancy.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -1044,7 +1043,7 @@ public synchronized long position(final TopicPartition
partition) {
}
@Test
- public void
shouldUseRequestTimeoutPlusTaskTimeoutInPollDuringRestoreAndFailIfNoDataReturned()
{
+ public void
shouldUsePollMsPlusRequestTimeoutInPollDuringRestoreAndTimeoutWhenNoProgressDuringRestore()
{
Review comment:
After the refactoring who we handle `task.timeout.ms` during global
restore, this test and the next test did cover the exact some thing. So I
removed one test to avoid redundancy. (The diff is a little weird as it cut's
out in the middle...)
--
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
