mjsax commented on code in PR #15414: URL: https://github.com/apache/kafka/pull/15414#discussion_r1527393295
########## streams/src/test/java/org/apache/kafka/streams/integration/GlobalStateReprocessTest.java: ########## @@ -0,0 +1,205 @@ +/* + * 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.LongSerializer; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.StoreQueryParameters; +import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster; +import org.apache.kafka.streams.integration.utils.IntegrationTestUtils; +import org.apache.kafka.streams.kstream.Consumed; +import org.apache.kafka.streams.processor.api.ContextualProcessor; +import org.apache.kafka.streams.processor.api.ProcessorSupplier; +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.QueryableStoreTypes; +import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; +import org.apache.kafka.streams.state.Stores; +import org.apache.kafka.streams.state.internals.KeyValueStoreBuilder; +import org.apache.kafka.test.TestUtils; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInfo; +import org.junit.jupiter.api.Timeout; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Properties; + +import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; + +@Timeout(600) +@Tag("integration") +public class GlobalStateReprocessTest { + private static final int NUM_BROKERS = 1; + private static final Properties BROKER_CONFIG; + + static { + BROKER_CONFIG = new Properties(); + BROKER_CONFIG.put("transaction.state.log.replication.factor", (short) 1); + BROKER_CONFIG.put("transaction.state.log.min.isr", 1); + } + + public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS, BROKER_CONFIG); + + @BeforeAll + public static void startCluster() throws IOException { + CLUSTER.start(); + } + + @AfterAll + public static void closeCluster() { + CLUSTER.stop(); + } + + + private final MockTime mockTime = CLUSTER.time; + private final String globalStore = "globalStore"; + private StreamsBuilder builder; + private Properties streamsConfiguration; + private KafkaStreams kafkaStreams; + private String globalStoreTopic; + + + @BeforeEach + public void before(final TestInfo testInfo) throws Exception { + builder = new StreamsBuilder(); + + createTopics(); + streamsConfiguration = new Properties(); + final String safeTestName = safeUniqueTestName(testInfo); + streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" + safeTestName); + streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); + streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath()); + streamsConfiguration.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0); + streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L); + + final KeyValueStoreBuilder<String, Long> storeBuilder = new KeyValueStoreBuilder<>( + Stores.persistentKeyValueStore(globalStore), + Serdes.String(), + Serdes.Long(), + mockTime); + + final ProcessorSupplier<String, Long, Void, Void> processorSupplier; + processorSupplier = () -> new ContextualProcessor<String, Long, Void, Void>() { + @Override + public void process(final Record<String, Long> record) { + final KeyValueStore<String, Long> stateStore = + context().getStateStore(storeBuilder.name()); + stateStore.put( + record.key() + "- this is the right value.", + record.value() + ); + } + }; + + builder.addGlobalStore( + storeBuilder, + globalStoreTopic, + Consumed.with(Serdes.String(), Serdes.Long()), + processorSupplier + ); + } + + @AfterEach + public void after() throws Exception { + if (kafkaStreams != null) { + kafkaStreams.close(); + } + IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration); + } + + @Test + public void shouldReprocessWithUserProvidedStore() throws Exception { + kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration); + populateTopics(globalStoreTopic); + + kafkaStreams.start(); + + TestUtils.waitForCondition( + () -> !storeContents(kafkaStreams).isEmpty(), + 30000, + "Has not processed record within 30 seconds"); + + assertThat(storeContents(kafkaStreams).get(0), containsString("- this is the right value.")); + + + kafkaStreams.close(); + kafkaStreams.cleanUp(); + + kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration); + kafkaStreams.start(); + + TestUtils.waitForCondition( + () -> !storeContents(kafkaStreams).isEmpty(), + 30000, + "Has not processed record within 30 seconds"); + + assertThat(storeContents(kafkaStreams).get(0), containsString("- this is the right value.")); + } + + private void createTopics() throws Exception { + globalStoreTopic = "global-store-topic"; + CLUSTER.createTopic(globalStoreTopic); + } + + private void populateTopics(final String topicName) throws Exception { + IntegrationTestUtils.produceKeyValuesSynchronously( + topicName, + Arrays.asList( + new KeyValue<>("A", 1L), + new KeyValue<>("B", 2L), + new KeyValue<>("C", 3L), + new KeyValue<>("D", 4L)), Review Comment: Why are we writing 4 records? Seems one would be sufficient given what assertions the test makes? ########## streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java: ########## @@ -236,6 +252,91 @@ private List<TopicPartition> topicPartitionsForStore(final StateStore store) { } return topicPartitions; } + @SuppressWarnings("unchecked") + private void reprocessState(final List<TopicPartition> topicPartitions, + final Map<TopicPartition, Long> highWatermarks, + final InternalTopologyBuilder.ReprocessFactory reprocessFactory, + final String storeName) { + final Processor source = reprocessFactory.processorSupplier().get(); + source.init(globalProcessorContext); + + for (final TopicPartition topicPartition : topicPartitions) { + long currentDeadline = NO_DEADLINE; + + globalConsumer.assign(Collections.singletonList(topicPartition)); + long offset; + final Long checkpoint = checkpointFileCache.get(topicPartition); + if (checkpoint != null) { + globalConsumer.seek(topicPartition, checkpoint); + offset = checkpoint; + } else { + globalConsumer.seekToBeginning(Collections.singletonList(topicPartition)); + offset = getGlobalConsumerOffset(topicPartition); + } + final Long highWatermark = highWatermarks.get(topicPartition); + stateRestoreListener.onRestoreStart(topicPartition, storeName, offset, highWatermark); + + long restoreCount = 0L; + + while (offset < highWatermark) { + // we add `request.timeout.ms` to `poll.ms` because `poll.ms` might be too short + // to give a fetch request a fair chance to actually complete and we don't want to + // start `task.timeout.ms` too early + // + // TODO with https://issues.apache.org/jira/browse/KAFKA-10315 we can just call + // `poll(pollMS)` without adding the request timeout and do a more precise + // timeout handling + final ConsumerRecords<byte[], byte[]> records = globalConsumer.poll(pollMsPlusRequestTimeout); + if (records.isEmpty()) { + currentDeadline = maybeUpdateDeadlineOrThrow(currentDeadline); + } else { + currentDeadline = NO_DEADLINE; + } + + for (final ConsumerRecord<byte[], byte[]> record : records.records(topicPartition)) { + final ProcessorRecordContext recordContext = + new ProcessorRecordContext( + record.timestamp(), + record.offset(), + record.partition(), + record.topic(), + record.headers()); + globalProcessorContext.setRecordContext(recordContext); + + try { + if (record.key() != null) { + source.process(new Record<>( + reprocessFactory.keyDeserializer().deserialize(record.topic(), record.key()), + reprocessFactory.valueDeserializer().deserialize(record.topic(), record.value()), + record.timestamp(), + record.headers())); + restoreCount++; + } + } catch (final Exception deserializationException) { + handleDeserializationFailure( Review Comment: Wondering if we should create `new RecordDeserializer` and keep everything encapsulated in there without the need to refactor and create ` handleDeserializationFailure`? ########## streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java: ########## @@ -68,40 +68,49 @@ ConsumerRecord<Object, Object> deserialize(final ProcessorContext<?, ?> processo Optional.empty() ); } catch (final Exception deserializationException) { - final DeserializationExceptionHandler.DeserializationHandlerResponse response; - try { - response = deserializationExceptionHandler.handle( - (InternalProcessorContext<?, ?>) processorContext, - rawRecord, - deserializationException); - } catch (final Exception fatalUserException) { - log.error( - "Deserialization error callback failed after deserialization error for record {}", - rawRecord, - deserializationException); - throw new StreamsException("Fatal user code error in deserialization error callback", fatalUserException); - } + handleDeserializationFailure(deserializationExceptionHandler, processorContext, deserializationException, rawRecord, log, droppedRecordsSensor); + return null; Review Comment: Semantics of `return null` is unclear w/o looking into `handleDeserializationFailure` -- we should at least put a comment `// 'handleDeserializationFailure' would either throw or swallow -- if we swallow we need to skip the record by returning 'null'` (or something like this)? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
