You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "cadonna (via GitHub)" <gi...@apache.org> on 2023/12/18 21:13:43 UTC
[PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
cadonna opened a new pull request, #15044:
URL: https://github.com/apache/kafka/pull/15044
Kafka Streams checkpoints the wrong offset when a task is closed during restoration. If under exactly-once processing guarantees a TaskCorruptedException happens, the affected task is closed dirty, its state content is wiped out and the task is re-initialized. If during the following restoration the task is closed cleanly, the task writes the offsets that it stores in its record collector to the checkpoint file. Those offsets are the offsets that the task wrote to the changelog topics. In other words, the task writes the end offsets of its changelog topics to the checkpoint file. Consequently, when the task is initialized again on the same Streams client, the checkpoint file is read and the task assumes it is fully restored although the records between the last offsets the task restored before closing clean and the end offset of the changelog topics are missing locally.
The fix is to clear the offsets in the record collector on close.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #15044:
URL: https://github.com/apache/kafka/pull/15044#issuecomment-1864411254
LGTM, thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1432578710
##########
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java:
##########
@@ -1313,7 +1313,6 @@ private static <K, V> List<ConsumerRecord<K, V>> readRecords(final String topic,
final int maxMessages) {
final List<ConsumerRecord<K, V>> consumerRecords;
consumer.subscribe(singletonList(topic));
- System.out.println("Got assignment:" + consumer.assignment());
Review Comment:
I discovered this and removed it, because I guess it is a left-over from some other PR.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1431219204
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -377,6 +377,11 @@ public void closeDirty() {
streamsProducer.abortTransaction();
}
+ close();
+ }
+
+ private void close() {
+ offsets.clear();
Review Comment:
I am afraid I do not understand. Method `removeAllProducedSensors()` is only called in `closeClean()` but not in `closedDirty()`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1431214126
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
Review Comment:
No, it is not essential. Actually it is just to put some load on restoration.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1431220982
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
+ value1.append("A");
+ }
+ final String valueStr1 = value1.toString();
+ final List<KeyValue<Integer, String>> recordBatch1 = IntStream.range(startKey, endKey).mapToObj(i -> KeyValue.pair(i, valueStr1)).collect(Collectors.toList());
+ IntegrationTestUtils.produceKeyValuesSynchronously(MULTI_PARTITION_INPUT_TOPIC,
+ recordBatch1,
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ IntegerSerializer.class,
+ StringSerializer.class),
+ CLUSTER.time);
+
+ final StoreBuilder<KeyValueStore<Integer, String>> stateStore = Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore(stateStoreName),
+ Serdes.Integer(),
+ Serdes.String()).withCachingEnabled();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicBoolean throwException = new AtomicBoolean(false);
+ final TaskId task00 = new TaskId(0, 0);
+ final AtomicLong restoredOffsetsForPartition0 = new AtomicLong(0);
+ final Topology topology = new Topology();
+ topology
+ .addSource("source", MULTI_PARTITION_INPUT_TOPIC)
+ .addProcessor("processor", () -> new Processor<Integer, String, Integer, String>() {
+ KeyValueStore<Integer, String> stateStore;
+ org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context;
+
+ @Override
+ public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context) {
+ Processor.super.init(context);
+ this.context = context;
+ stateStore = context.getStateStore(stateStoreName);
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ context.recordMetadata().ifPresent(recordMetadata -> {
+ if (recordMetadata.partition() == 0) {
+ if (throwException.compareAndSet(true, false)) {
+ throw new TaskCorruptedException(Collections.singleton(task00));
+ }
+ stateStore.put(record.key(), record.value());
+ } else {
+ stateStore.put(record.key(), record.value());
+ if (restoredOffsetsForPartition0.get() > 0) {
+ latch.countDown();
+ }
+ }
+ });
+ }
+
+ @Override
+ public void close() {
+ Processor.super.close();
+ }
+ }, "source")
+ .addStateStore(stateStore, "processor")
+ .addSink("sink", MULTI_PARTITION_OUTPUT_TOPIC, "processor");
+
+ final KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfiguration);
+ kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
+ @Override
+ public void onRestoreStart(final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset,
+ final long endingOffset) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Starting offset: " + startingOffset);
+ }
+ }
+ @Override
+ public void onBatchRestored(final TopicPartition topicPartition,
+ final String storeName,
+ final long batchEndOffset,
+ final long numRestored) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Batch end offset: " + batchEndOffset);
+ restoredOffsetsForPartition0.set(batchEndOffset);
+ }
+ }
+ @Override
+ public void onRestoreEnd(final TopicPartition topicPartition,
+ final String storeName,
+ final long totalRestored) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Total restored: " + totalRestored);
+ }
+ }
+ });
+ startApplicationAndWaitUntilRunning(Collections.singletonList(kafkaStreams), Duration.ofSeconds(60));
+ ensureCommittedRecordsInTopic(
+ applicationId + "-" + stateStoreName + "-changelog",
+ 2000,
+ StringDeserializer.class,
+ StringDeserializer.class
+ );
+ throwException.set(true);
Review Comment:
Why is that a risk?
If you are referring to the `else`-branch in the processor, I changed that in the last commit.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1430656569
##########
checkstyle/import-control.xml:
##########
@@ -408,6 +408,7 @@
<allow pkg="com.fasterxml.jackson" />
<allow pkg="kafka.utils" />
<allow pkg="org.apache.zookeeper" />
+ <allow pkg="org.apache.log4j" />
Review Comment:
I needed to add this to enable https://github.com/apache/kafka/pull/15044/files#r1430656200
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1431209859
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
+ value1.append("A");
+ }
+ final String valueStr1 = value1.toString();
+ final List<KeyValue<Integer, String>> recordBatch1 = IntStream.range(startKey, endKey).mapToObj(i -> KeyValue.pair(i, valueStr1)).collect(Collectors.toList());
+ IntegrationTestUtils.produceKeyValuesSynchronously(MULTI_PARTITION_INPUT_TOPIC,
+ recordBatch1,
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ IntegerSerializer.class,
+ StringSerializer.class),
+ CLUSTER.time);
+
+ final StoreBuilder<KeyValueStore<Integer, String>> stateStore = Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore(stateStoreName),
+ Serdes.Integer(),
+ Serdes.String()).withCachingEnabled();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicBoolean throwException = new AtomicBoolean(false);
+ final TaskId task00 = new TaskId(0, 0);
+ final AtomicLong restoredOffsetsForPartition0 = new AtomicLong(0);
+ final Topology topology = new Topology();
+ topology
+ .addSource("source", MULTI_PARTITION_INPUT_TOPIC)
+ .addProcessor("processor", () -> new Processor<Integer, String, Integer, String>() {
+ KeyValueStore<Integer, String> stateStore;
+ org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context;
+
+ @Override
+ public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context) {
+ Processor.super.init(context);
+ this.context = context;
+ stateStore = context.getStateStore(stateStoreName);
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ context.recordMetadata().ifPresent(recordMetadata -> {
+ if (recordMetadata.partition() == 0) {
+ if (throwException.compareAndSet(true, false)) {
+ throw new TaskCorruptedException(Collections.singleton(task00));
+ }
+ stateStore.put(record.key(), record.value());
+ } else {
+ stateStore.put(record.key(), record.value());
+ if (restoredOffsetsForPartition0.get() > 0) {
+ latch.countDown();
+ }
+ }
+ });
+ }
+
+ @Override
+ public void close() {
+ Processor.super.close();
+ }
+ }, "source")
+ .addStateStore(stateStore, "processor")
+ .addSink("sink", MULTI_PARTITION_OUTPUT_TOPIC, "processor");
+
+ final KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfiguration);
+ kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
+ @Override
+ public void onRestoreStart(final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset,
+ final long endingOffset) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Starting offset: " + startingOffset);
+ }
+ }
+ @Override
+ public void onBatchRestored(final TopicPartition topicPartition,
+ final String storeName,
+ final long batchEndOffset,
+ final long numRestored) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Batch end offset: " + batchEndOffset);
+ restoredOffsetsForPartition0.set(batchEndOffset);
+ }
+ }
+ @Override
+ public void onRestoreEnd(final TopicPartition topicPartition,
+ final String storeName,
+ final long totalRestored) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Total restored: " + totalRestored);
+ }
+ }
+ });
+ startApplicationAndWaitUntilRunning(Collections.singletonList(kafkaStreams), Duration.ofSeconds(60));
+ ensureCommittedRecordsInTopic(
+ applicationId + "-" + stateStoreName + "-changelog",
+ 2000,
+ StringDeserializer.class,
+ StringDeserializer.class
+ );
+ throwException.set(true);
Review Comment:
Is there a risk here that I may process all 30k keys before I manage to throw the exception?
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
+ value1.append("A");
+ }
+ final String valueStr1 = value1.toString();
+ final List<KeyValue<Integer, String>> recordBatch1 = IntStream.range(startKey, endKey).mapToObj(i -> KeyValue.pair(i, valueStr1)).collect(Collectors.toList());
+ IntegrationTestUtils.produceKeyValuesSynchronously(MULTI_PARTITION_INPUT_TOPIC,
+ recordBatch1,
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ IntegerSerializer.class,
+ StringSerializer.class),
+ CLUSTER.time);
+
+ final StoreBuilder<KeyValueStore<Integer, String>> stateStore = Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore(stateStoreName),
+ Serdes.Integer(),
+ Serdes.String()).withCachingEnabled();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicBoolean throwException = new AtomicBoolean(false);
+ final TaskId task00 = new TaskId(0, 0);
+ final AtomicLong restoredOffsetsForPartition0 = new AtomicLong(0);
+ final Topology topology = new Topology();
+ topology
+ .addSource("source", MULTI_PARTITION_INPUT_TOPIC)
+ .addProcessor("processor", () -> new Processor<Integer, String, Integer, String>() {
+ KeyValueStore<Integer, String> stateStore;
+ org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context;
+
+ @Override
+ public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context) {
+ Processor.super.init(context);
+ this.context = context;
+ stateStore = context.getStateStore(stateStoreName);
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ context.recordMetadata().ifPresent(recordMetadata -> {
+ if (recordMetadata.partition() == 0) {
+ if (throwException.compareAndSet(true, false)) {
+ throw new TaskCorruptedException(Collections.singleton(task00));
+ }
+ stateStore.put(record.key(), record.value());
+ } else {
+ stateStore.put(record.key(), record.value());
+ if (restoredOffsetsForPartition0.get() > 0) {
+ latch.countDown();
+ }
+ }
+ });
+ }
+
+ @Override
+ public void close() {
+ Processor.super.close();
+ }
+ }, "source")
+ .addStateStore(stateStore, "processor")
+ .addSink("sink", MULTI_PARTITION_OUTPUT_TOPIC, "processor");
+
+ final KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfiguration);
+ kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
+ @Override
+ public void onRestoreStart(final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset,
+ final long endingOffset) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Starting offset: " + startingOffset);
Review Comment:
Do we really want to println this?
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
+ value1.append("A");
+ }
+ final String valueStr1 = value1.toString();
+ final List<KeyValue<Integer, String>> recordBatch1 = IntStream.range(startKey, endKey).mapToObj(i -> KeyValue.pair(i, valueStr1)).collect(Collectors.toList());
+ IntegrationTestUtils.produceKeyValuesSynchronously(MULTI_PARTITION_INPUT_TOPIC,
+ recordBatch1,
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ IntegerSerializer.class,
+ StringSerializer.class),
+ CLUSTER.time);
+
+ final StoreBuilder<KeyValueStore<Integer, String>> stateStore = Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore(stateStoreName),
+ Serdes.Integer(),
+ Serdes.String()).withCachingEnabled();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicBoolean throwException = new AtomicBoolean(false);
+ final TaskId task00 = new TaskId(0, 0);
+ final AtomicLong restoredOffsetsForPartition0 = new AtomicLong(0);
+ final Topology topology = new Topology();
+ topology
+ .addSource("source", MULTI_PARTITION_INPUT_TOPIC)
+ .addProcessor("processor", () -> new Processor<Integer, String, Integer, String>() {
+ KeyValueStore<Integer, String> stateStore;
+ org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context;
+
+ @Override
+ public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context) {
+ Processor.super.init(context);
+ this.context = context;
+ stateStore = context.getStateStore(stateStoreName);
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ context.recordMetadata().ifPresent(recordMetadata -> {
+ if (recordMetadata.partition() == 0) {
+ if (throwException.compareAndSet(true, false)) {
+ throw new TaskCorruptedException(Collections.singleton(task00));
+ }
+ stateStore.put(record.key(), record.value());
+ } else {
+ stateStore.put(record.key(), record.value());
+ if (restoredOffsetsForPartition0.get() > 0) {
+ latch.countDown();
+ }
+ }
+ });
+ }
+
+ @Override
+ public void close() {
+ Processor.super.close();
+ }
+ }, "source")
+ .addStateStore(stateStore, "processor")
+ .addSink("sink", MULTI_PARTITION_OUTPUT_TOPIC, "processor");
+
+ final KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfiguration);
+ kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
+ @Override
+ public void onRestoreStart(final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset,
+ final long endingOffset) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Starting offset: " + startingOffset);
+ }
+ }
+ @Override
+ public void onBatchRestored(final TopicPartition topicPartition,
+ final String storeName,
+ final long batchEndOffset,
+ final long numRestored) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Batch end offset: " + batchEndOffset);
Review Comment:
Do we really want to println this?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -377,6 +377,11 @@ public void closeDirty() {
streamsProducer.abortTransaction();
}
+ close();
+ }
+
+ private void close() {
+ offsets.clear();
Review Comment:
I like having common logic for close clean / close dirty. Should we also move `removeAllProducedSensors` here?
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
+ value1.append("A");
+ }
+ final String valueStr1 = value1.toString();
+ final List<KeyValue<Integer, String>> recordBatch1 = IntStream.range(startKey, endKey).mapToObj(i -> KeyValue.pair(i, valueStr1)).collect(Collectors.toList());
+ IntegrationTestUtils.produceKeyValuesSynchronously(MULTI_PARTITION_INPUT_TOPIC,
+ recordBatch1,
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ IntegerSerializer.class,
+ StringSerializer.class),
+ CLUSTER.time);
+
+ final StoreBuilder<KeyValueStore<Integer, String>> stateStore = Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore(stateStoreName),
+ Serdes.Integer(),
+ Serdes.String()).withCachingEnabled();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicBoolean throwException = new AtomicBoolean(false);
+ final TaskId task00 = new TaskId(0, 0);
+ final AtomicLong restoredOffsetsForPartition0 = new AtomicLong(0);
+ final Topology topology = new Topology();
+ topology
+ .addSource("source", MULTI_PARTITION_INPUT_TOPIC)
+ .addProcessor("processor", () -> new Processor<Integer, String, Integer, String>() {
+ KeyValueStore<Integer, String> stateStore;
+ org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context;
+
+ @Override
+ public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context) {
+ Processor.super.init(context);
+ this.context = context;
+ stateStore = context.getStateStore(stateStoreName);
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ context.recordMetadata().ifPresent(recordMetadata -> {
+ if (recordMetadata.partition() == 0) {
+ if (throwException.compareAndSet(true, false)) {
+ throw new TaskCorruptedException(Collections.singleton(task00));
+ }
+ stateStore.put(record.key(), record.value());
+ } else {
+ stateStore.put(record.key(), record.value());
+ if (restoredOffsetsForPartition0.get() > 0) {
+ latch.countDown();
+ }
+ }
+ });
+ }
+
+ @Override
+ public void close() {
+ Processor.super.close();
+ }
+ }, "source")
+ .addStateStore(stateStore, "processor")
+ .addSink("sink", MULTI_PARTITION_OUTPUT_TOPIC, "processor");
+
+ final KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfiguration);
+ kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
+ @Override
+ public void onRestoreStart(final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset,
+ final long endingOffset) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Starting offset: " + startingOffset);
+ }
+ }
+ @Override
+ public void onBatchRestored(final TopicPartition topicPartition,
+ final String storeName,
+ final long batchEndOffset,
+ final long numRestored) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Batch end offset: " + batchEndOffset);
+ restoredOffsetsForPartition0.set(batchEndOffset);
+ }
+ }
+ @Override
+ public void onRestoreEnd(final TopicPartition topicPartition,
+ final String storeName,
+ final long totalRestored) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Total restored: " + totalRestored);
Review Comment:
Do we really want to println this?
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
Review Comment:
Is the size of the value essential to the test? In other words, "why?"
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1431243577
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
+ value1.append("A");
+ }
+ final String valueStr1 = value1.toString();
+ final List<KeyValue<Integer, String>> recordBatch1 = IntStream.range(startKey, endKey).mapToObj(i -> KeyValue.pair(i, valueStr1)).collect(Collectors.toList());
+ IntegrationTestUtils.produceKeyValuesSynchronously(MULTI_PARTITION_INPUT_TOPIC,
+ recordBatch1,
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ IntegerSerializer.class,
+ StringSerializer.class),
+ CLUSTER.time);
+
+ final StoreBuilder<KeyValueStore<Integer, String>> stateStore = Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore(stateStoreName),
+ Serdes.Integer(),
+ Serdes.String()).withCachingEnabled();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicBoolean throwException = new AtomicBoolean(false);
+ final TaskId task00 = new TaskId(0, 0);
+ final AtomicLong restoredOffsetsForPartition0 = new AtomicLong(0);
+ final Topology topology = new Topology();
+ topology
+ .addSource("source", MULTI_PARTITION_INPUT_TOPIC)
+ .addProcessor("processor", () -> new Processor<Integer, String, Integer, String>() {
+ KeyValueStore<Integer, String> stateStore;
+ org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context;
+
+ @Override
+ public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context) {
+ Processor.super.init(context);
+ this.context = context;
+ stateStore = context.getStateStore(stateStoreName);
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ context.recordMetadata().ifPresent(recordMetadata -> {
+ if (recordMetadata.partition() == 0) {
+ if (throwException.compareAndSet(true, false)) {
+ throw new TaskCorruptedException(Collections.singleton(task00));
+ }
+ stateStore.put(record.key(), record.value());
+ } else {
+ stateStore.put(record.key(), record.value());
+ if (restoredOffsetsForPartition0.get() > 0) {
+ latch.countDown();
+ }
+ }
+ });
+ }
+
+ @Override
+ public void close() {
+ Processor.super.close();
+ }
+ }, "source")
+ .addStateStore(stateStore, "processor")
+ .addSink("sink", MULTI_PARTITION_OUTPUT_TOPIC, "processor");
+
+ final KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfiguration);
+ kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
+ @Override
+ public void onRestoreStart(final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset,
+ final long endingOffset) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Starting offset: " + startingOffset);
+ }
+ }
+ @Override
+ public void onBatchRestored(final TopicPartition topicPartition,
+ final String storeName,
+ final long batchEndOffset,
+ final long numRestored) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Batch end offset: " + batchEndOffset);
+ restoredOffsetsForPartition0.set(batchEndOffset);
+ }
+ }
+ @Override
+ public void onRestoreEnd(final TopicPartition topicPartition,
+ final String storeName,
+ final long totalRestored) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Total restored: " + totalRestored);
+ }
+ }
+ });
+ startApplicationAndWaitUntilRunning(Collections.singletonList(kafkaStreams), Duration.ofSeconds(60));
+ ensureCommittedRecordsInTopic(
+ applicationId + "-" + stateStoreName + "-changelog",
+ 2000,
+ StringDeserializer.class,
+ StringDeserializer.class
+ );
+ throwException.set(true);
Review Comment:
OK, I see that there is a chance to not throw the exception in the `if`-branch of the processor. I added a second batch of records to add to the input topic after ensuring committed records in the changelog topic.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1430656200
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java:
##########
@@ -1228,6 +1290,7 @@ public void shouldNotThrowStreamsExceptionOnSubsequentCallIfASendFailsWithContin
try (final LogCaptureAppender logCaptureAppender =
LogCaptureAppender.createAndRegister(RecordCollectorImpl.class)) {
+ logCaptureAppender.setThreshold(Level.INFO);
Review Comment:
I added this because otherwise the result of the test depends on the configured log level.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna merged PR #15044:
URL: https://github.com/apache/kafka/pull/15044
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1431241782
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
Review Comment:
OK, I removed the string value and put in place a integer value instead.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1430656965
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -377,6 +377,11 @@ public void closeDirty() {
streamsProducer.abortTransaction();
}
+ close();
+ }
+
+ private void close() {
+ offsets.clear();
Review Comment:
This is the actual fix.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1432717987
##########
checkstyle/suppressions.xml:
##########
@@ -230,7 +230,7 @@
<!-- Streams tests -->
<suppress checks="ClassFanOutComplexity"
- files="(RecordCollectorTest|StreamsPartitionAssignorTest|StreamThreadTest|StreamTaskTest|TaskManagerTest|TopologyTestDriverTest|KafkaStreamsTest).java"/>
+ files="(RecordCollectorTest|StreamsPartitionAssignorTest|StreamThreadTest|StreamTaskTest|TaskManagerTest|TopologyTestDriverTest|KafkaStreamsTest|EosIntegrationTest).java"/>
Review Comment:
I could have probably solved this checkstyle issue by moving the test to a separate file but I think it makes sense to keep it in `EosIntegrationTest` to avoid starting an additional embedded Kafka.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1432581521
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -1006,12 +1177,40 @@ private List<KeyValue<Long, Long>> readResult(final String topic,
// read uncommitted
return IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
- TestUtils.consumerConfig(CLUSTER.bootstrapServers(), LongDeserializer.class, LongDeserializer.class),
+ TestUtils.consumerConfig(CLUSTER.bootstrapServers(), keyDeserializer, valueDeserializer),
topic,
numberOfRecords
);
}
+ private <K, V> void ensureCommittedRecordsInTopicPartition(final String topic,
Review Comment:
I added this method to specifically verify if the partition to verify contains committed records because I saw flaky test failures where the changelog topic of the partition to verify was empty. If the changelog topic is empty, the latch never counts down, the Streams client never closes and the test runs into the test timeout.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16017: Checkpoint restored offsets instead of written offsets [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15044:
URL: https://github.com/apache/kafka/pull/15044#discussion_r1431211947
##########
streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -789,6 +806,158 @@ public void shouldWriteLatestOffsetsToCheckpointOnShutdown() throws Exception {
verifyOffsetsAreInCheckpoint(1);
}
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterEnabled() throws Exception {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(true);
+ }
+
+ @Test
+ public void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoringStateUpdaterDisabled() throws Exception {
+ if (!processingThreadsEnabled) {
+ shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(false);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ private void shouldCheckpointRestoredOffsetsWhenClosingCleanDuringRestoring(final boolean stateUpdaterEnabled) throws Exception {
+ if (!eosConfig.equals(StreamsConfig.EXACTLY_ONCE) && !eosConfig.equals(StreamsConfig.EXACTLY_ONCE_V2)) {
+ return;
+ }
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig);
+ streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(applicationId).getPath());
+ streamsConfiguration.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
+ streamsConfiguration.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
+ final String stateStoreName = "stateStore";
+
+ purgeLocalStreamsState(streamsConfiguration);
+
+ final int startKey = 1;
+ final int endKey = 30001;
+ final int valueSize = 1000;
+ final StringBuilder value1 = new StringBuilder(valueSize);
+ for (int i = 0; i < valueSize; ++i) {
+ value1.append("A");
+ }
+ final String valueStr1 = value1.toString();
+ final List<KeyValue<Integer, String>> recordBatch1 = IntStream.range(startKey, endKey).mapToObj(i -> KeyValue.pair(i, valueStr1)).collect(Collectors.toList());
+ IntegrationTestUtils.produceKeyValuesSynchronously(MULTI_PARTITION_INPUT_TOPIC,
+ recordBatch1,
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ IntegerSerializer.class,
+ StringSerializer.class),
+ CLUSTER.time);
+
+ final StoreBuilder<KeyValueStore<Integer, String>> stateStore = Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore(stateStoreName),
+ Serdes.Integer(),
+ Serdes.String()).withCachingEnabled();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final AtomicBoolean throwException = new AtomicBoolean(false);
+ final TaskId task00 = new TaskId(0, 0);
+ final AtomicLong restoredOffsetsForPartition0 = new AtomicLong(0);
+ final Topology topology = new Topology();
+ topology
+ .addSource("source", MULTI_PARTITION_INPUT_TOPIC)
+ .addProcessor("processor", () -> new Processor<Integer, String, Integer, String>() {
+ KeyValueStore<Integer, String> stateStore;
+ org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context;
+
+ @Override
+ public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context) {
+ Processor.super.init(context);
+ this.context = context;
+ stateStore = context.getStateStore(stateStoreName);
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ context.recordMetadata().ifPresent(recordMetadata -> {
+ if (recordMetadata.partition() == 0) {
+ if (throwException.compareAndSet(true, false)) {
+ throw new TaskCorruptedException(Collections.singleton(task00));
+ }
+ stateStore.put(record.key(), record.value());
+ } else {
+ stateStore.put(record.key(), record.value());
+ if (restoredOffsetsForPartition0.get() > 0) {
+ latch.countDown();
+ }
+ }
+ });
+ }
+
+ @Override
+ public void close() {
+ Processor.super.close();
+ }
+ }, "source")
+ .addStateStore(stateStore, "processor")
+ .addSink("sink", MULTI_PARTITION_OUTPUT_TOPIC, "processor");
+
+ final KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfiguration);
+ kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
+ @Override
+ public void onRestoreStart(final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset,
+ final long endingOffset) {
+ if (topicPartition.partition() == 0) {
+ System.out.println("Restore listener - Starting offset: " + startingOffset);
Review Comment:
No, removed those. Those were leftovers from debugging.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org