mjsax commented on code in PR #17198:
URL: https://github.com/apache/kafka/pull/17198#discussion_r1828373616
##########
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();
Review Comment:
Why would we use `var` here? We know the type and thus should use `int`?
`var` should only be used for lambdas in which we might not know the type.
--
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
