guozhangwang commented on code in PR #13369:
URL: https://github.com/apache/kafka/pull/13369#discussion_r1139441498
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -590,12 +590,13 @@ void recycle() {
}
}
- void transitionTaskType(final TaskType newType) {
+ void transitionTaskType(final TaskType newType, final LogContext
logContext) {
Review Comment:
This is a piggy-backed fix: I realized that when we recycle tasks, the log
context should also be updated to change the prefix. Otherwise it would be very
confusing in the logs since a recycled `standby` task would still print as
`active` task in the prefix and vice versa.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -846,7 +846,7 @@ private void initializeChangelogs(final Map<TaskId, Task>
tasks,
changelogMetadata.restoreEndOffset = Math.min(endOffset,
committedOffset);
- log.debug("End offset for changelog {} initialized as {}.",
partition, changelogMetadata.restoreEndOffset);
+ log.info("End offset for changelog {} initialized as {}.",
partition, changelogMetadata.restoreEndOffset);
Review Comment:
I found this log line super helpful for restoration tasks, while at the same
time it only prints once per partition during restoration so not a huge cost.
##########
streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSMultiRebalanceIntegrationTest.java:
##########
@@ -0,0 +1,304 @@
+/*
+ * 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.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StoreQueryParameters;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.Transformer;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.QueryableStoreTypes;
+import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.test.IntegrationTest;
+import org.apache.kafka.test.TestUtils;
+import org.hamcrest.Matchers;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static
org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+import static org.apache.kafka.test.TestUtils.waitForCondition;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+/**
+ * An integration test to verify EOS properties when using Caching and Standby
replicas
+ * while tasks are being redistributed after re-balancing event.
+ * The intent is not that this test should be merged into the repo but only
provided for evidence on how to reproduce.
+ * One test fail and two test pass reliably on an i7-8750H CPU @ 2.20GHz × 12
with 32 GiB Memory
+ */
+@Category(IntegrationTest.class)
+@SuppressWarnings("deprecation")
+public class StandbyTaskEOSMultiRebalanceIntegrationTest {
Review Comment:
The credit of this integration test goes to @chorslev . I tried to make it a
bit shorter.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java:
##########
@@ -1200,4 +1227,6 @@ private static class ConverterStore extends
MockKeyValueStore implements Timesta
super(name, persistent);
}
}
+
+ interface CachingStore extends CachedStateStore, StateStore { }
Review Comment:
This is due to limitation of easy mock, Mockito would not need this hack.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java:
##########
@@ -280,6 +280,13 @@ synchronized void close(final String namespace) {
}
}
+ synchronized void clear(final String namespace) {
+ final NamedCache cleared = caches.get(namespace);
+ if (cleared != null) {
+ cleared.clear();
Review Comment:
Ack
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
