KAFKA-8602: Fix bug in stand-by task creation

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -495,7 +495,7 @@ StandbyTask createTask(final Consumer<byte[], byte[]> consumer,
 
             final ProcessorTopology topology = builder.build(taskId.topicGroupId);
 
-            if (!topology.stateStores().isEmpty()) {
+            if (!topology.stateStores().isEmpty() && !topology.storeToChangelogTopic().isEmpty()) {
                 return new StandbyTask(
                     taskId,
                     partitions,

streams/src/test/java/org/apache/kafka/streams/integration/StandByTaskCreationIntegrationTest.java

@@ -0,0 +1,139 @@
+/*
+ * 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.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KafkaStreams.State;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.kstream.Consumed;
+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.StoreBuilder;
+import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.test.IntegrationTest;
+import org.apache.kafka.test.TestUtils;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.Properties;
+
+@Category({IntegrationTest.class})
+public class StandByTaskCreationIntegrationTest {
+
+    private static final int NUM_BROKERS = 1;
+
+    @ClassRule
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);
+
+    private static final String INPUT_TOPIC = "input-topic";
+    private static final String OUTPUT_TOPIC = "output-topic";
+
+    @BeforeClass
+    public static void createTopics() throws InterruptedException {
+        CLUSTER.createTopic(INPUT_TOPIC, 2, 1);
+        CLUSTER.createTopic(OUTPUT_TOPIC, 2, 1);
+    }
+
+    private Properties streamsConfiguration() {
+        final String applicationId = "testApp";
+        final Properties streamsConfiguration = new Properties();
+        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+        streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+        streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+        streamsConfiguration.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
+        streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+        streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        return streamsConfiguration;
+    }
+
+    @Test
+    public void shouldNotCreateAnyStandByTasksForStateStoreWithLoggingDisabled() throws InterruptedException {
+        final StreamsBuilder builder = new StreamsBuilder();
+
+        final String stateStoreName = "myTransformState";
+        final StoreBuilder<KeyValueStore<Integer, Integer>> keyValueStoreBuilder =
+            Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore(stateStoreName),
+                                        Serdes.Integer(),
+                                        Serdes.Integer()).withLoggingDisabled();
+        builder.addStateStore(keyValueStoreBuilder);
+
+        builder.stream(INPUT_TOPIC, Consumed.with(Serdes.Integer(), Serdes.Integer()))
+            .transform(() -> new Transformer<Integer, Integer, KeyValue<Integer, Integer>>() {
+                private KeyValueStore<Integer, Integer> state;
+
+                @SuppressWarnings("unchecked")
+                @Override
+                public void init(final ProcessorContext context) {
+                    state = (KeyValueStore<Integer, Integer>) context.getStateStore(stateStoreName);
+                }
+
+                @Override
+                public KeyValue<Integer, Integer> transform(final Integer key, final Integer value) {
+                    state.putIfAbsent(key, value);
+                    final KeyValue<Integer, Integer> result = new KeyValue<>(key, value);
+                    return result;
+                }
+
+                @Override
+                public void close() {}
+            }, stateStoreName)
+            .to(OUTPUT_TOPIC);
+
+        final Topology topology = builder.build();
+        final KafkaStreams client1 = new KafkaStreams(topology, streamsConfiguration());
+        final KafkaStreams client2 = new KafkaStreams(topology, streamsConfiguration());
+
+        final boolean[] client1IsOk = {false}; // has to be a final array, otherwise flag cannot be modified in lambda
+        client1.setStateListener((newState, oldState) -> {
+            if (newState == State.RUNNING &&
+                client1.localThreadsMetadata().iterator().next().standbyTasks().isEmpty()) {
+
+                client1IsOk[0] = true;
+            }
+        });
+        final boolean[] client2IsOk = {false}; // has to be a final array, otherwise flag cannot be modified in lambda
+        client2.setStateListener((newState, oldState) -> {
+            if (newState == State.RUNNING &&
+                client2.localThreadsMetadata().iterator().next().standbyTasks().isEmpty()) {
+
+                client2IsOk[0] = true;
+            }
+        });
+
+        client1.start();
+        client2.start();
+
+        TestUtils.waitForCondition(
+            () -> client1IsOk[0] && client2IsOk[0],
+            30 * 1000,
+            "At least one client is not in state RUNNING or has a stand-by task");
+
+        client1.close();
+        client2.close();
+    }
+}

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.InvalidOffsetException;
 import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.common.Cluster;
@@ -59,8 +60,10 @@
 import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
 import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
 import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.StoreBuilder;
 import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
 import org.apache.kafka.test.MockClientSupplier;
+import org.apache.kafka.test.MockKeyValueStoreBuilder;
 import org.apache.kafka.test.MockProcessor;
 import org.apache.kafka.test.MockStateRestoreListener;
 import org.apache.kafka.test.MockTimestampExtractor;
@@ -70,6 +73,7 @@
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+import org.slf4j.Logger;
 
 import java.io.File;
 import java.time.Duration;
@@ -92,6 +96,8 @@
 import static org.apache.kafka.streams.processor.internals.AbstractStateManager.CHECKPOINT_FILE_NAME;
 import static org.apache.kafka.streams.processor.internals.StreamThread.getSharedAdminClientId;
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -1057,6 +1063,69 @@ public void shouldUpdateStandbyTask() throws Exception {
         assertEquals(0, thread.standbyRecords().size());
     }
 
+    @Test
+    public void shouldCreateStandbyTask() {
+        final MockProcessor mockProcessor = new MockProcessor();
+        internalTopologyBuilder.addSource(null, "source1", null, null, null, topic1);
+        internalTopologyBuilder.addProcessor("processor1", () -> mockProcessor, "source1");
+        internalTopologyBuilder.addStateStore(new MockKeyValueStoreBuilder("myStore", true), "processor1");
+        final StreamThread.StandbyTaskCreator standbyTaskCreator = createStandbyTaskCreator(internalTopologyBuilder);
+
+        final StandbyTask standbyTask = standbyTaskCreator.createTask(
+            new MockConsumer<>(OffsetResetStrategy.EARLIEST),
+            new TaskId(1, 2),
+            Collections.emptySet());
+
+        assertThat(standbyTask, not(nullValue()));
+    }
+
+    @Test
+    public void shouldNotCreateStandbyTaskWithoutStateStores() {
+        final MockProcessor mockProcessor = new MockProcessor();
+        internalTopologyBuilder.addSource(null, "source1", null, null, null, topic1);
+        internalTopologyBuilder.addProcessor("processor1", () -> mockProcessor, "source1");
+        final StreamThread.StandbyTaskCreator standbyTaskCreator = createStandbyTaskCreator(internalTopologyBuilder);
+
+        final StandbyTask standbyTask = standbyTaskCreator.createTask(
+            new MockConsumer<>(OffsetResetStrategy.EARLIEST),
+            new TaskId(1, 2),
+            Collections.emptySet());
+
+        assertThat(standbyTask, nullValue());
+    }
+
+    @Test
+    public void shouldNotCreateStandbyTaskIfStateStoresHaveLoggingDisabled() {
+        final MockProcessor mockProcessor = new MockProcessor();
+        internalTopologyBuilder.addSource(null, "source1", null, null, null, topic1);
+        internalTopologyBuilder.addProcessor("processor1", () -> mockProcessor, "source1");
+        final StoreBuilder storeBuilder = new MockKeyValueStoreBuilder("myStore", true);
+        storeBuilder.withLoggingDisabled();
+        internalTopologyBuilder.addStateStore(storeBuilder, "processor1");
+        final StreamThread.StandbyTaskCreator standbyTaskCreator = createStandbyTaskCreator(internalTopologyBuilder);
+
+        final StandbyTask standbyTask = standbyTaskCreator.createTask(
+            new MockConsumer<>(OffsetResetStrategy.EARLIEST),
+            new TaskId(1, 2),
+            Collections.emptySet());
+
+        assertThat(standbyTask, nullValue());
+    }
+
+    private StreamThread.StandbyTaskCreator createStandbyTaskCreator(final InternalTopologyBuilder internalTopologyBuilder) {
+        final LogContext logContext = new LogContext("test");
+        final Logger log = logContext.logger(StreamThreadTest.class);
+        final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(metrics, clientId);
+        return new StreamThread.StandbyTaskCreator(
+            internalTopologyBuilder,
+            config,
+            streamsMetrics,
+            stateDirectory,
+            new MockChangelogReader(),
+            mockTime,
+            log);
+    }
+
     @Test
     public void shouldPunctuateActiveTask() {
         final List<Long> punctuatedStreamTime = new ArrayList<>();