diff --git a/docs/streams/upgrade-guide.html b/docs/streams/upgrade-guide.html
index 14544077403cd..0f819cb384f92 100644
--- a/docs/streams/upgrade-guide.html
+++ b/docs/streams/upgrade-guide.html
@@ -133,6 +133,14 @@
+
+ The Processor API now support so-called read-only state stores, added via
+ KIP-813.
+ These stores don't have a dedicated changelog topic, but use their source topic for fault-tolerance,
+ simlar to KTables with source-topic optimization enabled.
+
+
We added a new method to KafkaStreams, namely KafkaStreams#setStandbyUpdateListener() in
diff --git a/streams/src/main/java/org/apache/kafka/streams/Topology.java b/streams/src/main/java/org/apache/kafka/streams/Topology.java
index d9c810aa42599..d1f0d1eb8bc8e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/Topology.java
+++ b/streams/src/main/java/org/apache/kafka/streams/Topology.java
@@ -738,6 +738,88 @@ public synchronized Topology addStateStore(final StoreBuilder storeBuilder,
return this;
}
+ /**
+ * Adds a read-only {@link StateStore} to the topology.
+ *
+ * A read-only {@link StateStore} does not create a dedicated changelog topic but uses it's input topic as
+ * changelog; thus, the used topic should be configured with log compaction.
+ *
+ * The auto.offset.reset property will be set to earliest for this topic.
+ *
+ * The provided {@link ProcessorSupplier} will be used to create a processor for all messages received
+ * from the given topic. This processor should contain logic to keep the {@link StateStore} up-to-date.
+ *
+ * @param storeBuilder user defined store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
+ * @param timestampExtractor the stateless timestamp extractor used for this source,
+ * if not specified the default extractor defined in the configs will be used
+ * @param keyDeserializer the {@link Deserializer} to deserialize keys with
+ * @param valueDeserializer the {@link Deserializer} to deserialize values with
+ * @param topic the topic to source the data from
+ * @param processorName the name of the {@link ProcessorSupplier}
+ * @param stateUpdateSupplier the instance of {@link ProcessorSupplier}
+ * @return itself
+ * @throws TopologyException if the processor of state is already registered
+ */
+ public synchronized Topology addReadOnlyStateStore(final StoreBuilder storeBuilder,
+ final String sourceName,
+ final TimestampExtractor timestampExtractor,
+ final Deserializer keyDeserializer,
+ final Deserializer valueDeserializer,
+ final String topic,
+ final String processorName,
+ final ProcessorSupplier stateUpdateSupplier) {
+ storeBuilder.withLoggingDisabled();
+
+ internalTopologyBuilder.addSource(AutoOffsetReset.EARLIEST, sourceName, timestampExtractor, keyDeserializer, valueDeserializer, topic);
+ internalTopologyBuilder.addProcessor(processorName, stateUpdateSupplier, sourceName);
+ internalTopologyBuilder.addStateStore(storeBuilder, processorName);
+ internalTopologyBuilder.connectSourceStoreAndTopic(storeBuilder.name(), topic);
+
+ return this;
+ }
+
+ /**
+ * Adds a read-only {@link StateStore} to the topology.
+ *
+ * A read-only {@link StateStore} does not create a dedicated changelog topic but uses it's input topic as
+ * changelog; thus, the used topic should be configured with log compaction.
+ *
+ * The auto.offset.reset property will be set to earliest for this topic.
+ *
+ * The provided {@link ProcessorSupplier} will be used to create a processor for all messages received
+ * from the given topic. This processor should contain logic to keep the {@link StateStore} up-to-date.
+ * The default {@link TimestampExtractor} as specified in the {@link StreamsConfig config} is used.
+ *
+ * @param storeBuilder user defined store builder
+ * @param sourceName name of the {@link SourceNode} that will be automatically added
+ * @param keyDeserializer the {@link Deserializer} to deserialize keys with
+ * @param valueDeserializer the {@link Deserializer} to deserialize values with
+ * @param topic the topic to source the data from
+ * @param processorName the name of the {@link ProcessorSupplier}
+ * @param stateUpdateSupplier the instance of {@link ProcessorSupplier}
+ * @return itself
+ * @throws TopologyException if the processor of state is already registered
+ */
+ public synchronized Topology addReadOnlyStateStore(final StoreBuilder storeBuilder,
+ final String sourceName,
+ final Deserializer keyDeserializer,
+ final Deserializer valueDeserializer,
+ final String topic,
+ final String processorName,
+ final ProcessorSupplier stateUpdateSupplier) {
+ return addReadOnlyStateStore(
+ storeBuilder,
+ sourceName,
+ null,
+ keyDeserializer,
+ valueDeserializer,
+ topic,
+ processorName,
+ stateUpdateSupplier
+ );
+ }
+
/**
* Adds a global {@link StateStore} to the topology.
* The {@link StateStore} sources its data from all partitions of the provided input topic.
diff --git a/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
index 5fdf5c220cfa2..ceecbc3bc2178 100644
--- a/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
@@ -38,6 +38,7 @@
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.SubtopologyDescription;
import org.apache.kafka.streams.processor.internals.ProcessorTopology;
+import org.apache.kafka.streams.processor.internals.StoreFactory;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.SessionStore;
import org.apache.kafka.streams.state.StoreBuilder;
@@ -58,6 +59,7 @@
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
+import org.mockito.internal.util.collections.Sets;
import java.time.Duration;
import java.util.Arrays;
@@ -337,7 +339,7 @@ public void shouldNotAllowToAddStoreWithSameNameAndDifferentInstance() {
mockStoreBuilder();
topology.addStateStore(storeBuilder);
- final StoreBuilder otherStoreBuilder = mock(StoreBuilder.class);
+ final StoreBuilder otherStoreBuilder = mock(StoreBuilder.class);
when(otherStoreBuilder.name()).thenReturn("store");
when(otherStoreBuilder.logConfig()).thenReturn(Collections.emptyMap());
when(otherStoreBuilder.loggingEnabled()).thenReturn(false);
@@ -2313,7 +2315,7 @@ private TopologyDescription.Sink addSink(final String sinkName,
topology.addSink(sinkName, sinkTopic, null, null, null, parentNames);
final TopologyDescription.Sink expectedSinkNode =
- new InternalTopologyBuilder.Sink(sinkName, sinkTopic);
+ new InternalTopologyBuilder.Sink<>(sinkName, sinkTopic);
for (final TopologyDescription.Node parent : parents) {
((InternalTopologyBuilder.AbstractNode) parent).addSuccessor(expectedSinkNode);
@@ -2351,6 +2353,63 @@ private void addGlobalStoreToTopologyAndExpectedDescription(final String globalS
expectedDescription.addGlobalStore(expectedGlobalStore);
}
+ @Test
+ public void readOnlyStateStoresShouldHaveTheirOwnSubTopology() {
+ final String sourceName = "source";
+ final String storeName = "store";
+ final String topicName = "topic";
+ final String processorName = "processor";
+
+ final KeyValueStoreBuilder storeBuilder = mock(KeyValueStoreBuilder.class);
+ when(storeBuilder.name()).thenReturn(storeName);
+ topology.addReadOnlyStateStore(
+ storeBuilder,
+ sourceName,
+ null,
+ null,
+ null,
+ topicName,
+ processorName,
+ new MockProcessorSupplier<>());
+
+ final TopologyDescription.Source expectedSource = new InternalTopologyBuilder.Source(sourceName, Sets.newSet(topicName), null);
+ final TopologyDescription.Processor expectedProcessor = new InternalTopologyBuilder.Processor(processorName, Sets.newSet(storeName));
+
+ ((InternalTopologyBuilder.AbstractNode) expectedSource).addSuccessor(expectedProcessor);
+ ((InternalTopologyBuilder.AbstractNode) expectedProcessor).addPredecessor(expectedSource);
+
+ final Set allNodes = new HashSet<>();
+ allNodes.add(expectedSource);
+ allNodes.add(expectedProcessor);
+ expectedDescription.addSubtopology(new SubtopologyDescription(0, allNodes));
+
+ assertThat(topology.describe(), equalTo(expectedDescription));
+ assertThat(topology.describe().hashCode(), equalTo(expectedDescription.hashCode()));
+ }
+
+ @Test
+ public void readOnlyStateStoresShouldNotLog() {
+ final String sourceName = "source";
+ final String storeName = "store";
+ final String topicName = "topic";
+ final String processorName = "processor";
+
+ final KeyValueStoreBuilder storeBuilder = mock(KeyValueStoreBuilder.class);
+ when(storeBuilder.name()).thenReturn(storeName);
+ topology.addReadOnlyStateStore(
+ storeBuilder,
+ sourceName,
+ null,
+ null,
+ null,
+ topicName,
+ processorName,
+ new MockProcessorSupplier<>());
+
+ final StoreFactory stateStoreFactory = topology.internalTopologyBuilder.stateStores().get(storeName);
+ assertThat(stateStoreFactory.loggingEnabled(), equalTo(false));
+ }
+
private TopologyConfig overrideDefaultStore(final String defaultStore) {
final Properties topologyOverrides = new Properties();
// change default store as in-memory
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
index d79631f8f7a0c..7989de0e76bb7 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
@@ -123,8 +123,8 @@ public class EosIntegrationTest {
public Timeout globalTimeout = Timeout.seconds(600);
private static final Logger LOG = LoggerFactory.getLogger(EosIntegrationTest.class);
private static final int NUM_BROKERS = 3;
- private static final int MAX_POLL_INTERVAL_MS = 5 * 1000;
- private static final int MAX_WAIT_TIME_MS = 60 * 1000;
+ private static final int MAX_POLL_INTERVAL_MS = 30_1000;
+ private static final int MAX_WAIT_TIME_MS = 120_1000;
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(
NUM_BROKERS,
@@ -403,7 +403,7 @@ public void shouldNotViolateEosIfOneTaskFails() throws Exception {
// -> the failure only kills one thread
// after fail over, we should read 40 committed records (even if 50 record got written)
- try (final KafkaStreams streams = getKafkaStreams("dummy", false, "appDir", 2, eosConfig, MAX_POLL_INTERVAL_MS)) {
+ try (final KafkaStreams streams = getKafkaStreams("dummy", false, "appDir", 2, eosConfig)) {
startApplicationAndWaitUntilRunning(streams);
final List> committedDataBeforeFailure = prepareData(0L, 10L, 0L, 1L);
@@ -511,7 +511,7 @@ public void shouldNotViolateEosIfOneTaskFailsWithState() throws Exception {
// We need more processing time under "with state" situation, so increasing the max.poll.interval.ms
// to avoid unexpected rebalance during test, which will cause unexpected fail over triggered
- try (final KafkaStreams streams = getKafkaStreams("dummy", true, "appDir", 2, eosConfig, 3 * MAX_POLL_INTERVAL_MS)) {
+ try (final KafkaStreams streams = getKafkaStreams("dummy", true, "appDir", 2, eosConfig)) {
startApplicationAndWaitUntilRunning(streams);
final List> committedDataBeforeFailure = prepareData(0L, 10L, 0L, 1L);
@@ -624,12 +624,12 @@ public void shouldNotViolateEosIfOneTaskGetsFencedUsingIsolatedAppInstances() th
// -> the stall only affects one thread and should trigger a rebalance
// after rebalancing, we should read 40 committed records (even if 50 record got written)
//
- // afterwards, the "stalling" thread resumes, and another rebalance should get triggered
+ // afterward, the "stalling" thread resumes, and another rebalance should get triggered
// we write the remaining 20 records and verify to read 60 result records
try (
- final KafkaStreams streams1 = getKafkaStreams("streams1", false, "appDir1", 1, eosConfig, MAX_POLL_INTERVAL_MS);
- final KafkaStreams streams2 = getKafkaStreams("streams2", false, "appDir2", 1, eosConfig, MAX_POLL_INTERVAL_MS)
+ final KafkaStreams streams1 = getKafkaStreams("streams1", false, "appDir1", 1, eosConfig);
+ final KafkaStreams streams2 = getKafkaStreams("streams2", false, "appDir2", 1, eosConfig)
) {
startApplicationAndWaitUntilRunning(streams1);
startApplicationAndWaitUntilRunning(streams2);
@@ -778,7 +778,7 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
final List> writtenData = prepareData(0L, 10, 0L, 1L);
final List> expectedResult = computeExpectedResult(writtenData);
- try (final KafkaStreams streams = getKafkaStreams("streams", true, "appDir", 1, eosConfig, MAX_POLL_INTERVAL_MS)) {
+ try (final KafkaStreams streams = getKafkaStreams("streams", true, "appDir", 1, eosConfig)) {
writeInputData(writtenData);
startApplicationAndWaitUntilRunning(streams);
@@ -1004,8 +1004,7 @@ private KafkaStreams getKafkaStreams(final String dummyHostName,
final boolean withState,
final String appDir,
final int numberOfStreamsThreads,
- final String eosConfig,
- final int maxPollIntervalMs) {
+ final String eosConfig) {
commitRequested = new AtomicInteger(0);
errorInjected = new AtomicBoolean(false);
stallInjected = new AtomicBoolean(false);
@@ -1112,9 +1111,8 @@ public void close() { }
properties.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), (int) commitIntervalMs);
properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.METADATA_MAX_AGE_CONFIG), "1000");
properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), "earliest");
- properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG), maxPollIntervalMs);
- properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG), maxPollIntervalMs - 1);
- properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG), maxPollIntervalMs);
+ properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG), MAX_POLL_INTERVAL_MS - 1);
+ properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG), MAX_POLL_INTERVAL_MS);
properties.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
properties.put(StreamsConfig.STATE_DIR_CONFIG, stateTmpDir + appDir);
properties.put(StreamsConfig.APPLICATION_SERVER_CONFIG, dummyHostName + ":2142");
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
index 812da30074775..b41e0c3d29f2f 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
@@ -28,6 +28,7 @@
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.common.serialization.StringDeserializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Utils;
@@ -155,8 +156,8 @@ private Properties props(final Properties extraProperties) {
streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
streamsConfiguration.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId).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.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class);
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class);
streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000L);
streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
streamsConfiguration.putAll(extraProperties);
@@ -196,8 +197,8 @@ public void shouldRestoreNullRecord() throws Exception {
final Properties streamsConfiguration = StreamsTestUtils.getStreamsConfig(
applicationId,
CLUSTER.bootstrapServers(),
- Serdes.Integer().getClass().getName(),
- Serdes.ByteArray().getClass().getName(),
+ Serdes.IntegerSerde.class.getName(),
+ Serdes.BytesSerde.class.getName(),
props);
CLUSTER.createTopics(inputTopic);
@@ -249,7 +250,63 @@ public void shouldRestoreNullRecord() throws Exception {
@ParameterizedTest
@MethodSource("parameters")
- public void shouldRestoreStateFromSourceTopic(final boolean stateUpdaterEnabled) throws Exception {
+ public void shouldRestoreStateFromSourceTopicForReadOnlyStore(final boolean stateUpdaterEnabled) throws Exception {
+ final AtomicInteger numReceived = new AtomicInteger(0);
+ final Topology topology = new Topology();
+
+ final Properties props = props(stateUpdaterEnabled);
+
+ // restoring from 1000 to 4000 (committed), and then process from 4000 to 5000 on each of the two partitions
+ final int offsetLimitDelta = 1000;
+ final int offsetCheckpointed = 1000;
+ createStateForRestoration(inputStream, 0);
+ setCommittedOffset(inputStream, offsetLimitDelta);
+
+ final StateDirectory stateDirectory = new StateDirectory(new StreamsConfig(props), new MockTime(), true, false);
+ // note here the checkpointed offset is the last processed record's offset, so without control message we should write this offset - 1
+ new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(new TaskId(0, 0)), ".checkpoint"))
+ .write(Collections.singletonMap(new TopicPartition(inputStream, 0), (long) offsetCheckpointed - 1));
+ new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(new TaskId(0, 1)), ".checkpoint"))
+ .write(Collections.singletonMap(new TopicPartition(inputStream, 1), (long) offsetCheckpointed - 1));
+
+ final CountDownLatch startupLatch = new CountDownLatch(1);
+ final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ topology.addReadOnlyStateStore(
+ Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore("store"),
+ new Serdes.IntegerSerde(),
+ new Serdes.StringSerde()
+ ),
+ "readOnlySource",
+ new IntegerDeserializer(),
+ new StringDeserializer(),
+ inputStream,
+ "readOnlyProcessor",
+ () -> new ReadOnlyStoreProcessor(numReceived, offsetLimitDelta, shutdownLatch)
+ );
+
+ kafkaStreams = new KafkaStreams(topology, props);
+ kafkaStreams.setStateListener((newState, oldState) -> {
+ if (newState == KafkaStreams.State.RUNNING && oldState == KafkaStreams.State.REBALANCING) {
+ startupLatch.countDown();
+ }
+ });
+
+ final AtomicLong restored = new AtomicLong(0);
+ kafkaStreams.setGlobalStateRestoreListener(new TrackingStateRestoreListener(restored));
+ kafkaStreams.start();
+
+ assertTrue(startupLatch.await(30, TimeUnit.SECONDS));
+ assertThat(restored.get(), equalTo((long) numberOfKeys - offsetLimitDelta * 2 - offsetCheckpointed * 2));
+
+ assertTrue(shutdownLatch.await(30, TimeUnit.SECONDS));
+ assertThat(numReceived.get(), equalTo(offsetLimitDelta * 2));
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ public void shouldRestoreStateFromSourceTopicForGlobalTable(final boolean stateUpdaterEnabled) throws Exception {
final AtomicInteger numReceived = new AtomicInteger(0);
final StreamsBuilder builder = new StreamsBuilder();
@@ -265,9 +322,9 @@ public void shouldRestoreStateFromSourceTopic(final boolean stateUpdaterEnabled)
final StateDirectory stateDirectory = new StateDirectory(new StreamsConfig(props), new MockTime(), true, false);
// note here the checkpointed offset is the last processed record's offset, so without control message we should write this offset - 1
new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(new TaskId(0, 0)), ".checkpoint"))
- .write(Collections.singletonMap(new TopicPartition(inputStream, 0), (long) offsetCheckpointed - 1));
+ .write(Collections.singletonMap(new TopicPartition(inputStream, 0), (long) offsetCheckpointed - 1));
new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(new TaskId(0, 1)), ".checkpoint"))
- .write(Collections.singletonMap(new TopicPartition(inputStream, 1), (long) offsetCheckpointed - 1));
+ .write(Collections.singletonMap(new TopicPartition(inputStream, 1), (long) offsetCheckpointed - 1));
final CountDownLatch startupLatch = new CountDownLatch(1);
final CountDownLatch shutdownLatch = new CountDownLatch(1);
@@ -288,22 +345,7 @@ public void shouldRestoreStateFromSourceTopic(final boolean stateUpdaterEnabled)
});
final AtomicLong restored = new AtomicLong(0);
- kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
- @Override
- public void onRestoreStart(final TopicPartition topicPartition, final String storeName, final long startingOffset, final long endingOffset) {
-
- }
-
- @Override
- public void onBatchRestored(final TopicPartition topicPartition, final String storeName, final long batchEndOffset, final long numRestored) {
-
- }
-
- @Override
- public void onRestoreEnd(final TopicPartition topicPartition, final String storeName, final long totalRestored) {
- restored.addAndGet(totalRestored);
- }
- });
+ kafkaStreams.setGlobalStateRestoreListener(new TrackingStateRestoreListener(restored));
kafkaStreams.start();
assertTrue(startupLatch.await(30, TimeUnit.SECONDS));
@@ -332,9 +374,9 @@ public void shouldRestoreStateFromChangelogTopic(final boolean stateUpdaterEnabl
final StateDirectory stateDirectory = new StateDirectory(new StreamsConfig(props), new MockTime(), true, false);
// note here the checkpointed offset is the last processed record's offset, so without control message we should write this offset - 1
new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(new TaskId(0, 0)), ".checkpoint"))
- .write(Collections.singletonMap(new TopicPartition(changelog, 0), (long) offsetCheckpointed - 1));
+ .write(Collections.singletonMap(new TopicPartition(changelog, 0), (long) offsetCheckpointed - 1));
new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(new TaskId(0, 1)), ".checkpoint"))
- .write(Collections.singletonMap(new TopicPartition(changelog, 1), (long) offsetCheckpointed - 1));
+ .write(Collections.singletonMap(new TopicPartition(changelog, 1), (long) offsetCheckpointed - 1));
final CountDownLatch startupLatch = new CountDownLatch(1);
final CountDownLatch shutdownLatch = new CountDownLatch(1);
@@ -355,22 +397,7 @@ public void shouldRestoreStateFromChangelogTopic(final boolean stateUpdaterEnabl
});
final AtomicLong restored = new AtomicLong(0);
- kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
- @Override
- public void onRestoreStart(final TopicPartition topicPartition, final String storeName, final long startingOffset, final long endingOffset) {
-
- }
-
- @Override
- public void onBatchRestored(final TopicPartition topicPartition, final String storeName, final long batchEndOffset, final long numRestored) {
-
- }
-
- @Override
- public void onRestoreEnd(final TopicPartition topicPartition, final String storeName, final long totalRestored) {
- restored.addAndGet(totalRestored);
- }
- });
+ kafkaStreams.setGlobalStateRestoreListener(new TrackingStateRestoreListener(restored));
kafkaStreams.start();
assertTrue(startupLatch.await(30, TimeUnit.SECONDS));
@@ -386,10 +413,12 @@ public void shouldSuccessfullyStartWhenLoggingDisabled(final boolean stateUpdate
final StreamsBuilder builder = new StreamsBuilder();
final KStream stream = builder.stream(inputStream);
- stream.groupByKey()
- .reduce(
- (value1, value2) -> value1 + value2,
- Materialized.>as("reduce-store").withLoggingDisabled());
+ stream
+ .groupByKey()
+ .reduce(
+ Integer::sum,
+ Materialized.>as("reduce-store").withLoggingDisabled()
+ );
final CountDownLatch startupLatch = new CountDownLatch(1);
kafkaStreams = new KafkaStreams(builder.build(), props(stateUpdaterEnabled));
@@ -821,4 +850,30 @@ private void waitForTransitionTo(final Set observed, final K
() -> "Client did not transition to " + state + " on time. Observed transitions: " + observed
);
}
+
+ private static class ReadOnlyStoreProcessor implements Processor {
+ private final AtomicInteger numReceived;
+ private final int offsetLimitDelta;
+ private final CountDownLatch shutdownLatch;
+ KeyValueStore store;
+
+ public ReadOnlyStoreProcessor(final AtomicInteger numReceived, final int offsetLimitDelta, final CountDownLatch shutdownLatch) {
+ this.numReceived = numReceived;
+ this.offsetLimitDelta = offsetLimitDelta;
+ this.shutdownLatch = shutdownLatch;
+ }
+
+ @Override
+ public void init(final ProcessorContext context) {
+ store = context.getStateStore("store");
+ }
+
+ @Override
+ public void process(final Record record) {
+ store.put(record.key(), record.value());
+ if (numReceived.incrementAndGet() == offsetLimitDelta * 2) {
+ shutdownLatch.countDown();
+ }
+ }
+ }
}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
index 4f1d8d3d4266f..7c5734d7b96af 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
@@ -1518,6 +1518,15 @@ public static class TrackingStateRestoreListener implements StateRestoreListener
public final Map changelogToStartOffset = new ConcurrentHashMap<>();
public final Map changelogToEndOffset = new ConcurrentHashMap<>();
public final Map changelogToTotalNumRestored = new ConcurrentHashMap<>();
+ private final AtomicLong restored;
+
+ public TrackingStateRestoreListener() {
+ restored = null;
+ }
+
+ public TrackingStateRestoreListener(final AtomicLong restored) {
+ this.restored = restored;
+ }
@Override
public void onRestoreStart(final TopicPartition topicPartition,
@@ -1541,6 +1550,9 @@ public void onBatchRestored(final TopicPartition topicPartition,
public void onRestoreEnd(final TopicPartition topicPartition,
final String storeName,
final long totalRestored) {
+ if (restored != null) {
+ restored.addAndGet(totalRestored);
+ }
}
public long totalNumRestored() {
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/ReadOnlyStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/ReadOnlyStoreTest.java
new file mode 100644
index 0000000000000..a786c82bf25bd
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/ReadOnlyStoreTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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.processor;
+
+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.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.TestOutputTopic;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyTestDriver;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.Stores;
+import org.junit.Test;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+public class ReadOnlyStoreTest {
+
+ @Test
+ public void shouldConnectProcessorAndWriteDataToReadOnlyStore() {
+ final Topology topology = new Topology();
+ topology.addReadOnlyStateStore(
+ Stores.keyValueStoreBuilder(
+ Stores.inMemoryKeyValueStore("readOnlyStore"),
+ new Serdes.IntegerSerde(),
+ new Serdes.StringSerde()
+ ),
+ "readOnlySource",
+ new IntegerDeserializer(),
+ new StringDeserializer(),
+ "storeTopic",
+ "readOnlyProcessor",
+ () -> new Processor() {
+ KeyValueStore store;
+
+ @Override
+ public void init(final ProcessorContext context) {
+ store = context.getStateStore("readOnlyStore");
+ }
+ @Override
+ public void process(final Record record) {
+ store.put(record.key(), record.value());
+ }
+ }
+ );
+
+ topology.addSource("source", new IntegerDeserializer(), new StringDeserializer(), "inputTopic");
+ topology.addProcessor(
+ "processor",
+ () -> new Processor() {
+ ProcessorContext context;
+ KeyValueStore store;
+
+ @Override
+ public void init(final ProcessorContext context) {
+ this.context = context;
+ store = context.getStateStore("readOnlyStore");
+ }
+
+ @Override
+ public void process(final Record record) {
+ context.forward(record.withValue(
+ record.value() + " -- " + store.get(record.key())
+ ));
+ }
+ },
+ "source"
+ );
+ topology.connectProcessorAndStateStores("processor", "readOnlyStore");
+ topology.addSink("sink", "outputTopic", new IntegerSerializer(), new StringSerializer(), "processor");
+
+ try (final TopologyTestDriver driver = new TopologyTestDriver(topology)) {
+ final TestInputTopic readOnlyStoreTopic =
+ driver.createInputTopic("storeTopic", new IntegerSerializer(), new StringSerializer());
+ final TestInputTopic input =
+ driver.createInputTopic("inputTopic", new IntegerSerializer(), new StringSerializer());
+ final TestOutputTopic output =
+ driver.createOutputTopic("outputTopic", new IntegerDeserializer(), new StringDeserializer());
+
+ readOnlyStoreTopic.pipeInput(1, "foo");
+ readOnlyStoreTopic.pipeInput(2, "bar");
+
+ input.pipeInput(1, "bar");
+ input.pipeInput(2, "foo");
+
+ final KeyValueStore store = driver.getKeyValueStore("readOnlyStore");
+
+ try (final KeyValueIterator it = store.all()) {
+ final List> storeContent = new LinkedList<>();
+ it.forEachRemaining(storeContent::add);
+
+ final List> expectedResult = new LinkedList<>();
+ expectedResult.add(KeyValue.pair(1, "foo"));
+ expectedResult.add(KeyValue.pair(2, "bar"));
+
+ assertThat(storeContent, equalTo(expectedResult));
+ }
+
+ final List> expectedResult = new LinkedList<>();
+ expectedResult.add(KeyValue.pair(1, "bar -- foo"));
+ expectedResult.add(KeyValue.pair(2, "foo -- bar"));
+
+ assertThat(output.readKeyValuesToList(), equalTo(expectedResult));
+ }
+ }
+}
diff --git a/streams/src/test/java/org/apache/kafka/test/MockProcessor.java b/streams/src/test/java/org/apache/kafka/test/MockProcessor.java
index 9766d1c0fe85c..8c1812ccf14a2 100644
--- a/streams/src/test/java/org/apache/kafka/test/MockProcessor.java
+++ b/streams/src/test/java/org/apache/kafka/test/MockProcessor.java
@@ -32,7 +32,6 @@
public class MockProcessor implements Processor {
private final MockApiProcessor delegate;
-
public MockProcessor(final PunctuationType punctuationType,
final long scheduleInterval) {
delegate = new MockApiProcessor<>(punctuationType, scheduleInterval);
@@ -43,12 +42,12 @@ public MockProcessor() {
}
@Override
- public void init(ProcessorContext context) {
+ public void init(final ProcessorContext context) {
delegate.init(context);
}
@Override
- public void process(Record record) {
+ public void process(final Record record) {
delegate.process(record);
}