fonsdant commented on code in PR #17198:
URL: https://github.com/apache/kafka/pull/17198#discussion_r1797755907
##########
streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java:
##########
@@ -329,26 +331,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 Integer key = record.key();
+ final Integer 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);
+ context.forward(record);
+ return;
}
if (store.get(key) != null) {
- return null;
+ store.delete(key);
Review Comment:
I have thought it would be the most similar replacement for return null.
--
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
