You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "fonsdant (via GitHub)" <gi...@apache.org> on 2024/09/25 01:45:18 UTC
[PR] [WIP] KAFKA-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
fonsdant opened a new pull request, #17266:
URL: https://github.com/apache/kafka/pull/17266
(no comment)
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1849415411
##########
streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java:
##########
@@ -44,6 +44,8 @@
* @param <VR> transformed value type
* @see ValueTransformerSupplier
* @see ValueTransformerWithKeySupplier
+ * @see KStream#transformValues(ValueTransformerSupplier, String...)
+ * @see KStream#transformValues(ValueTransformerWithKeySupplier, String...)
Review Comment:
This should be `KTable` right?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerWithKey.java:
##########
@@ -47,6 +47,8 @@
* @param <VR> transformed value type
* @see ValueTransformer
* @see ValueTransformerWithKeySupplier
+ * @see KStream#transformValues(ValueTransformerSupplier, String...)
+ * @see KStream#transformValues(ValueTransformerWithKeySupplier, String...)
Review Comment:
This should be `KTable` right?
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1849419753
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- @Test
- public void testTransform() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final org.apache.kafka.streams.kstream.ValueTransformerSupplier<Number, Integer> valueTransformerSupplier =
- () -> new org.apache.kafka.streams.kstream.ValueTransformer<Number, Integer>() {
- private int total = 0;
-
- @Override
- public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { }
-
- @Override
- public Integer transform(final Number value) {
- total += value.intValue();
- return total;
- }
-
- @Override
- public void close() { }
- };
-
- final int[] expectedKeys = {1, 10, 100, 1000};
-
- final KStream<Integer, Integer> stream;
- stream = builder.stream(topicName, Consumed.with(Serdes.Integer(), Serdes.Integer()));
- stream.transformValues(valueTransformerSupplier).process(supplier);
-
- try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
- for (final int expectedKey : expectedKeys) {
- final TestInputTopic<Integer, Integer> inputTopic =
- driver.createInputTopic(topicName, new IntegerSerializer(), new IntegerSerializer());
- inputTopic.pipeInput(expectedKey, expectedKey * 10, expectedKey / 2L);
- }
- }
- final KeyValueTimestamp[] expected = {new KeyValueTimestamp<>(1, 10, 0),
- new KeyValueTimestamp<>(10, 110, 5),
- new KeyValueTimestamp<>(100, 1110, 50),
- new KeyValueTimestamp<>(1000, 11110, 500)};
-
- assertArrayEquals(expected, supplier.theCapturedProcessor().processed().toArray());
- }
-
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
Thanks. SG.
But I am wondering if we can remove `KStreamTransformValuesTest.java` completely for this case?
What also implies that we can remove `KStreamTransformValues.java`?
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on PR #17266:
URL: https://github.com/apache/kafka/pull/17266#issuecomment-2479927469
I will push the docs update too in a while.
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1847525567
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- @Test
- public void testTransform() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final org.apache.kafka.streams.kstream.ValueTransformerSupplier<Number, Integer> valueTransformerSupplier =
- () -> new org.apache.kafka.streams.kstream.ValueTransformer<Number, Integer>() {
- private int total = 0;
-
- @Override
- public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { }
-
- @Override
- public Integer transform(final Number value) {
- total += value.intValue();
- return total;
- }
-
- @Override
- public void close() { }
- };
-
- final int[] expectedKeys = {1, 10, 100, 1000};
-
- final KStream<Integer, Integer> stream;
- stream = builder.stream(topicName, Consumed.with(Serdes.Integer(), Serdes.Integer()));
- stream.transformValues(valueTransformerSupplier).process(supplier);
-
- try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
- for (final int expectedKey : expectedKeys) {
- final TestInputTopic<Integer, Integer> inputTopic =
- driver.createInputTopic(topicName, new IntegerSerializer(), new IntegerSerializer());
- inputTopic.pipeInput(expectedKey, expectedKey * 10, expectedKey / 2L);
- }
- }
- final KeyValueTimestamp[] expected = {new KeyValueTimestamp<>(1, 10, 0),
- new KeyValueTimestamp<>(10, 110, 5),
- new KeyValueTimestamp<>(100, 1110, 50),
- new KeyValueTimestamp<>(1000, 11110, 500)};
-
- assertArrayEquals(expected, supplier.theCapturedProcessor().processed().toArray());
- }
-
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
I think they both are covered by `org.apache.kafka.streams.kstream.internals.KStreamImplTest#shouldProcessValues`, since `processValues` just has `FixedKeyProcessorSupplier` as processor supplier.
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1854862205
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- @Test
- public void testTransform() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final org.apache.kafka.streams.kstream.ValueTransformerSupplier<Number, Integer> valueTransformerSupplier =
- () -> new org.apache.kafka.streams.kstream.ValueTransformer<Number, Integer>() {
- private int total = 0;
-
- @Override
- public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { }
-
- @Override
- public Integer transform(final Number value) {
- total += value.intValue();
- return total;
- }
-
- @Override
- public void close() { }
- };
-
- final int[] expectedKeys = {1, 10, 100, 1000};
-
- final KStream<Integer, Integer> stream;
- stream = builder.stream(topicName, Consumed.with(Serdes.Integer(), Serdes.Integer()));
- stream.transformValues(valueTransformerSupplier).process(supplier);
-
- try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
- for (final int expectedKey : expectedKeys) {
- final TestInputTopic<Integer, Integer> inputTopic =
- driver.createInputTopic(topicName, new IntegerSerializer(), new IntegerSerializer());
- inputTopic.pipeInput(expectedKey, expectedKey * 10, expectedKey / 2L);
- }
- }
- final KeyValueTimestamp[] expected = {new KeyValueTimestamp<>(1, 10, 0),
- new KeyValueTimestamp<>(10, 110, 5),
- new KeyValueTimestamp<>(100, 1110, 50),
- new KeyValueTimestamp<>(1000, 11110, 500)};
-
- assertArrayEquals(expected, supplier.theCapturedProcessor().processed().toArray());
- }
-
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
> Following this approach, we could remove KStreamFlatTransformTest and KStreamFlatTransform too, couldn't we?
Maybe in PR [4/4] but not in this PR, right?
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1851155007
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- @Test
- public void testTransform() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final org.apache.kafka.streams.kstream.ValueTransformerSupplier<Number, Integer> valueTransformerSupplier =
- () -> new org.apache.kafka.streams.kstream.ValueTransformer<Number, Integer>() {
- private int total = 0;
-
- @Override
- public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { }
-
- @Override
- public Integer transform(final Number value) {
- total += value.intValue();
- return total;
- }
-
- @Override
- public void close() { }
- };
-
- final int[] expectedKeys = {1, 10, 100, 1000};
-
- final KStream<Integer, Integer> stream;
- stream = builder.stream(topicName, Consumed.with(Serdes.Integer(), Serdes.Integer()));
- stream.transformValues(valueTransformerSupplier).process(supplier);
-
- try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
- for (final int expectedKey : expectedKeys) {
- final TestInputTopic<Integer, Integer> inputTopic =
- driver.createInputTopic(topicName, new IntegerSerializer(), new IntegerSerializer());
- inputTopic.pipeInput(expectedKey, expectedKey * 10, expectedKey / 2L);
- }
- }
- final KeyValueTimestamp[] expected = {new KeyValueTimestamp<>(1, 10, 0),
- new KeyValueTimestamp<>(10, 110, 5),
- new KeyValueTimestamp<>(100, 1110, 50),
- new KeyValueTimestamp<>(1000, 11110, 500)};
-
- assertArrayEquals(expected, supplier.theCapturedProcessor().processed().toArray());
- }
-
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
@fonsdant Did you miss this comment?
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17266:
URL: https://github.com/apache/kafka/pull/17266#issuecomment-2495065870
Thanks for the PR. Merged to `trunk`.
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1847506802
##########
streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java:
##########
@@ -77,9 +75,6 @@ public interface ValueTransformer<V, VR> {
/**
* Transform the given value to a new value.
- * Additionally, any {@link StateStore} that is {@link KStream#transformValues(ValueTransformerSupplier, String...)
- * attached} to this operator can be accessed and modified arbitrarily (cf.
- * {@link ProcessorContext#getStateStore(String)}).
Review Comment:
Great idea!
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1847506107
##########
streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerWithKey.java:
##########
@@ -77,9 +75,6 @@ public interface ValueTransformerWithKey<K, V, VR> {
/**
* Transform the given [key and] value to a new value.
- * Additionally, any {@link StateStore} that is {@link KStream#transformValues(ValueTransformerWithKeySupplier, String...)
Review Comment:
As above.
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1847510563
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- @Test
- public void testTransform() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final org.apache.kafka.streams.kstream.ValueTransformerSupplier<Number, Integer> valueTransformerSupplier =
- () -> new org.apache.kafka.streams.kstream.ValueTransformer<Number, Integer>() {
- private int total = 0;
-
- @Override
- public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { }
-
- @Override
- public Integer transform(final Number value) {
- total += value.intValue();
- return total;
- }
-
- @Override
- public void close() { }
- };
-
- final int[] expectedKeys = {1, 10, 100, 1000};
-
- final KStream<Integer, Integer> stream;
- stream = builder.stream(topicName, Consumed.with(Serdes.Integer(), Serdes.Integer()));
- stream.transformValues(valueTransformerSupplier).process(supplier);
-
- try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
- for (final int expectedKey : expectedKeys) {
- final TestInputTopic<Integer, Integer> inputTopic =
- driver.createInputTopic(topicName, new IntegerSerializer(), new IntegerSerializer());
- inputTopic.pipeInput(expectedKey, expectedKey * 10, expectedKey / 2L);
- }
- }
- final KeyValueTimestamp[] expected = {new KeyValueTimestamp<>(1, 10, 0),
- new KeyValueTimestamp<>(10, 110, 5),
- new KeyValueTimestamp<>(100, 1110, 50),
- new KeyValueTimestamp<>(1000, 11110, 500)};
-
- assertArrayEquals(expected, supplier.theCapturedProcessor().processed().toArray());
- }
-
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
as above
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17266:
URL: https://github.com/apache/kafka/pull/17266#issuecomment-2484512795
Made a pass. Overall LGTM. A few comments:
> I will push the docs update too in a while.
Thx. PR [4/4] should remove `flatTransformValues` thought, right?
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1854863825
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- @Test
- public void testTransform() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final org.apache.kafka.streams.kstream.ValueTransformerSupplier<Number, Integer> valueTransformerSupplier =
- () -> new org.apache.kafka.streams.kstream.ValueTransformer<Number, Integer>() {
- private int total = 0;
-
- @Override
- public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { }
-
- @Override
- public Integer transform(final Number value) {
- total += value.intValue();
- return total;
- }
-
- @Override
- public void close() { }
- };
-
- final int[] expectedKeys = {1, 10, 100, 1000};
-
- final KStream<Integer, Integer> stream;
- stream = builder.stream(topicName, Consumed.with(Serdes.Integer(), Serdes.Integer()));
- stream.transformValues(valueTransformerSupplier).process(supplier);
-
- try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
- for (final int expectedKey : expectedKeys) {
- final TestInputTopic<Integer, Integer> inputTopic =
- driver.createInputTopic(topicName, new IntegerSerializer(), new IntegerSerializer());
- inputTopic.pipeInput(expectedKey, expectedKey * 10, expectedKey / 2L);
- }
- }
- final KeyValueTimestamp[] expected = {new KeyValueTimestamp<>(1, 10, 0),
- new KeyValueTimestamp<>(10, 110, 5),
- new KeyValueTimestamp<>(100, 1110, 50),
- new KeyValueTimestamp<>(1000, 11110, 500)};
-
- assertArrayEquals(expected, supplier.theCapturedProcessor().processed().toArray());
- }
-
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
> Following this approach, we could remove KStreamFlatTransformTest and KStreamFlatTransform too, couldn't we?
Maybe in PR [4/4], but not in this PR, right?
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1851851884
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- @Test
- public void testTransform() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final org.apache.kafka.streams.kstream.ValueTransformerSupplier<Number, Integer> valueTransformerSupplier =
- () -> new org.apache.kafka.streams.kstream.ValueTransformer<Number, Integer>() {
- private int total = 0;
-
- @Override
- public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { }
-
- @Override
- public Integer transform(final Number value) {
- total += value.intValue();
- return total;
- }
-
- @Override
- public void close() { }
- };
-
- final int[] expectedKeys = {1, 10, 100, 1000};
-
- final KStream<Integer, Integer> stream;
- stream = builder.stream(topicName, Consumed.with(Serdes.Integer(), Serdes.Integer()));
- stream.transformValues(valueTransformerSupplier).process(supplier);
-
- try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
- for (final int expectedKey : expectedKeys) {
- final TestInputTopic<Integer, Integer> inputTopic =
- driver.createInputTopic(topicName, new IntegerSerializer(), new IntegerSerializer());
- inputTopic.pipeInput(expectedKey, expectedKey * 10, expectedKey / 2L);
- }
- }
- final KeyValueTimestamp[] expected = {new KeyValueTimestamp<>(1, 10, 0),
- new KeyValueTimestamp<>(10, 110, 5),
- new KeyValueTimestamp<>(100, 1110, 50),
- new KeyValueTimestamp<>(1000, 11110, 500)};
-
- assertArrayEquals(expected, supplier.theCapturedProcessor().processed().toArray());
- }
-
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
@mjsax, oh, sorry! Yes, I have missed it. Thanks!
Makes sense to me too! I'm going to push the removal.
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1854862205
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- @Test
- public void testTransform() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final org.apache.kafka.streams.kstream.ValueTransformerSupplier<Number, Integer> valueTransformerSupplier =
- () -> new org.apache.kafka.streams.kstream.ValueTransformer<Number, Integer>() {
- private int total = 0;
-
- @Override
- public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { }
-
- @Override
- public Integer transform(final Number value) {
- total += value.intValue();
- return total;
- }
-
- @Override
- public void close() { }
- };
-
- final int[] expectedKeys = {1, 10, 100, 1000};
-
- final KStream<Integer, Integer> stream;
- stream = builder.stream(topicName, Consumed.with(Serdes.Integer(), Serdes.Integer()));
- stream.transformValues(valueTransformerSupplier).process(supplier);
-
- try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
- for (final int expectedKey : expectedKeys) {
- final TestInputTopic<Integer, Integer> inputTopic =
- driver.createInputTopic(topicName, new IntegerSerializer(), new IntegerSerializer());
- inputTopic.pipeInput(expectedKey, expectedKey * 10, expectedKey / 2L);
- }
- }
- final KeyValueTimestamp[] expected = {new KeyValueTimestamp<>(1, 10, 0),
- new KeyValueTimestamp<>(10, 110, 5),
- new KeyValueTimestamp<>(100, 1110, 50),
- new KeyValueTimestamp<>(1000, 11110, 500)};
-
- assertArrayEquals(expected, supplier.theCapturedProcessor().processed().toArray());
- }
-
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
> Following this approach, we could remove KStreamFlatTransformTest and KStreamFlatTransform too, couldn't we?
Maybe in PR [4/4] but not in this PR, right?
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #17266:
URL: https://github.com/apache/kafka/pull/17266
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1847510397
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
Do we have a corresponding test for `process(FixedKeyProcessor)` -- the comment says "needs to be migrated -- did this migration already happen?
--
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] [WIP] KAFKA-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17266:
URL: https://github.com/apache/kafka/pull/17266#issuecomment-2477701952
@fonsdant -- any updates?
--
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] [WIP] KAFKA-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on PR #17266:
URL: https://github.com/apache/kafka/pull/17266#issuecomment-2477744135
@mjsax, yes! I must be pushing some commits tomorrow. I am checking if there is need to refactor the tests.
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1847506525
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java:
##########
@@ -1209,48 +1207,6 @@ private <VO, VR> KStream<K, VR> doStreamTableJoin(final KTable<K, VO> table,
builder);
}
- @Override
- @Deprecated
- public <VR> KStream<K, VR> transformValues(final org.apache.kafka.streams.kstream.ValueTransformerSupplier<? super V, ? extends VR> valueTransformerSupplier,
- final String... stateStoreNames) {
- Objects.requireNonNull(valueTransformerSupplier, "valueTransformerSupplier can't be null");
- return doTransformValues(
- toValueTransformerWithKeySupplier(valueTransformerSupplier),
- NamedInternal.empty(),
- stateStoreNames);
- }
-
- @Override
- @Deprecated
- public <VR> KStream<K, VR> transformValues(final org.apache.kafka.streams.kstream.ValueTransformerSupplier<? super V, ? extends VR> valueTransformerSupplier,
- final Named named,
- final String... stateStoreNames) {
- Objects.requireNonNull(valueTransformerSupplier, "valueTransformerSupplier can't be null");
- Objects.requireNonNull(named, "named can't be null");
- return doTransformValues(
- toValueTransformerWithKeySupplier(valueTransformerSupplier),
- new NamedInternal(named),
- stateStoreNames);
- }
-
- @Override
- @Deprecated
- public <VR> KStream<K, VR> transformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerSupplier,
- final String... stateStoreNames) {
- Objects.requireNonNull(valueTransformerSupplier, "valueTransformerSupplier can't be null");
- return doTransformValues(valueTransformerSupplier, NamedInternal.empty(), stateStoreNames);
- }
-
- @Override
- @Deprecated
- public <VR> KStream<K, VR> transformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerSupplier,
- final Named named,
- final String... stateStoreNames) {
- Objects.requireNonNull(valueTransformerSupplier, "valueTransformerSupplier can't be null");
- Objects.requireNonNull(named, "named can't be null");
- return doTransformValues(valueTransformerSupplier, new NamedInternal(named), stateStoreNames);
- }
-
private <VR> KStream<K, VR> doTransformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerWithKeySupplier,
Review Comment:
This method should be unused now, and we can remove it too?
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1847505805
##########
streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java:
##########
@@ -77,9 +75,6 @@ public interface ValueTransformer<V, VR> {
/**
* Transform the given value to a new value.
- * Additionally, any {@link StateStore} that is {@link KStream#transformValues(ValueTransformerSupplier, String...)
- * attached} to this operator can be accessed and modified arbitrarily (cf.
- * {@link ProcessorContext#getStateStore(String)}).
Review Comment:
Should we keep this comment and link to `KTable#transformValues(...)` instead of `KStream` ?
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1850469033
##########
streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java:
##########
@@ -44,6 +44,8 @@
* @param <VR> transformed value type
* @see ValueTransformerSupplier
* @see ValueTransformerWithKeySupplier
+ * @see KStream#transformValues(ValueTransformerSupplier, String...)
+ * @see KStream#transformValues(ValueTransformerWithKeySupplier, String...)
Review Comment:
Yes! Sorry, I have missed this after `git reset`.
--
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-16339: [3/4 KStream#transformValues] Remove Deprecated "transformer" methods and classes [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on code in PR #17266:
URL: https://github.com/apache/kafka/pull/17266#discussion_r1851851884
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java:
##########
@@ -54,92 +44,6 @@ public class KStreamTransformValuesTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.Integer());
private InternalProcessorContext context = mock(InternalProcessorContext.class);
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
- @Test
- public void testTransform() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final org.apache.kafka.streams.kstream.ValueTransformerSupplier<Number, Integer> valueTransformerSupplier =
- () -> new org.apache.kafka.streams.kstream.ValueTransformer<Number, Integer>() {
- private int total = 0;
-
- @Override
- public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { }
-
- @Override
- public Integer transform(final Number value) {
- total += value.intValue();
- return total;
- }
-
- @Override
- public void close() { }
- };
-
- final int[] expectedKeys = {1, 10, 100, 1000};
-
- final KStream<Integer, Integer> stream;
- stream = builder.stream(topicName, Consumed.with(Serdes.Integer(), Serdes.Integer()));
- stream.transformValues(valueTransformerSupplier).process(supplier);
-
- try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
- for (final int expectedKey : expectedKeys) {
- final TestInputTopic<Integer, Integer> inputTopic =
- driver.createInputTopic(topicName, new IntegerSerializer(), new IntegerSerializer());
- inputTopic.pipeInput(expectedKey, expectedKey * 10, expectedKey / 2L);
- }
- }
- final KeyValueTimestamp[] expected = {new KeyValueTimestamp<>(1, 10, 0),
- new KeyValueTimestamp<>(10, 110, 5),
- new KeyValueTimestamp<>(100, 1110, 50),
- new KeyValueTimestamp<>(1000, 11110, 500)};
-
- assertArrayEquals(expected, supplier.theCapturedProcessor().processed().toArray());
- }
-
- @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
Review Comment:
@mjsax, oh, sorry! Yes, I have missed it. Thanks!
Makes sense to me too! I'm going to push the removal.
Following this approach, we could remove `KStreamFlatTransformTest` and `KStreamFlatTransform` too, couldn't we?
--
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