mjsax commented on code in PR #17198:
URL: https://github.com/apache/kafka/pull/17198#discussion_r1828375800
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java:
##########
@@ -329,26 +330,32 @@ private KafkaStreams buildWithDeduplicationTopology(final
String stateDirPath) {
Serdes.Integer())
);
builder.<Integer, Integer>stream(inputTopic)
- .transform(
- () -> new
org.apache.kafka.streams.kstream.Transformer<Integer, Integer,
KeyValue<Integer, Integer>>() {
+ .process(
+ () -> new Processor<Integer, Integer, Integer, Integer>() {
+ private ProcessorContext<Integer, Integer> context;
private KeyValueStore<Integer, Integer> store;
- @SuppressWarnings("unchecked")
@Override
- public void init(final ProcessorContext context) {
- store = (KeyValueStore<Integer, Integer>)
context.getStateStore(storeName);
+ public void init(final ProcessorContext<Integer, Integer>
context) {
+ this.context = context;
+ store = context.getStateStore(storeName);
}
@Override
- public KeyValue<Integer, Integer> transform(final Integer
key, final Integer value) {
+ public void process(final Record<Integer, Integer> record)
{
+ final var key = record.key();
+ final var value = record.value();
+
if (skipRecord.get()) {
// we only forward so we can verify the skipping
by reading the output topic
// the goal is skipping is to not modify the state
store
- return KeyValue.pair(key, value);
+ store.put(key, value);
+ context.forward(record);
+ return;
}
if (store.get(key) != null) {
- return null;
+ store.delete(key);
Review Comment:
Why do we add this line? Seem we just want to `return;` here to exit early?
--
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
