You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "UladzislauBlok (via GitHub)" <gi...@apache.org> on 2026/08/11 20:06:19 UTC
[PR] Suppress header-less partitioner deprecation [kafka]
UladzislauBlok opened a new pull request, #23134:
URL: https://github.com/apache/kafka/pull/23134
as a title
We could think about promoting some set of standard test partitioners into common util, but would be overkill IMO
--
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] MINOR: Suppress header-less partitioner deprecation [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23134:
URL: https://github.com/apache/kafka/pull/23134#discussion_r3792463580
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamRepartitionIntegrationTest.java:
##########
@@ -335,8 +336,14 @@ public void shouldRepartitionToMultiplePartitions(final String topologyOptimizat
final List<KeyValue<Integer, String>> expectedRecords = expectedRecordsOnRepartition.subList(3, 5);
class BroadcastingPartitioner implements StreamPartitioner<Integer, String> {
+ @SuppressWarnings("removal")
@Override
public Optional<Set<Integer>> partitions(final String topic, final Integer key, final String value, final int numPartitions) {
+ throw new AssertionError("Deprecated 4-argument partitions method was called instead of 5-argument method containing headers.");
Review Comment:
#22993 used a similar approach but it feels a bit verbose to me. Have we discussed adding a small helper to `org.apache.kafka.streams.utils.TestUtils` to simply build the partitioner?
```java
public static <K, V> StreamPartitioner<K, V> streamPartitioner(NexStreamPartitioner<K, V> partitioner) {
return new StreamPartitioner<>() {
@SuppressWarnings("removal")
@Override
public Optional<Set<Integer>> partitions(final String topic, final K key, final V value, final int numPartitions) {
throw new AssertionError("Deprecated 4-argument partitions method was called instead of 5-argument method containing headers.");
}
@Override
public Optional<Set<Integer>> partitions(final String topic, final K key, final V value, final Headers headers, final int numPartitions) {
return partitioner.partitions(topic, key, value, headers, numPartitions);
}
};
}
@FunctionalInterface
public interface NexStreamPartitioner<K, V> {
Optional<Set<Integer>> partitions(final String topic, final K key, final V value, final Headers headers, final int numPartitions);
}
```
```java
private final StreamPartitioner<String, Object> streamPartitioner =
TestUtils.streamPartitioner((topic, key, value, headers, numPartitions) ->
Optional.of(Collections.singleton(Integer.parseInt(key) % numPartitions)));
```
--
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] MINOR: Suppress header-less partitioner deprecation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23134:
URL: https://github.com/apache/kafka/pull/23134#discussion_r3762223463
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java:
##########
@@ -1371,4 +1366,23 @@ public long extract(final ConsumerRecord<Object, Object> record, final long part
return DEFAULT_TIMESTAMP;
}
}
+
+ private static class FixedPartitionPartitioner implements StreamPartitioner<Integer, Object> {
Review Comment:
```suggestion
private static class FixedPartitionPartitioner implements StreamPartitioner<String, String> {
```
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java:
##########
@@ -222,4 +224,23 @@ private Properties streamsConfiguration(final String safeTestName) {
config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
return config;
}
+
+ private static class FixedPartitionPartitioner implements StreamPartitioner<Integer, Object> {
Review Comment:
Seems we are duplicating code here (3x in this PR), plus already existing `FixedPartitionPartitioner` in `StreamsMetadataStateTest`).
Might be worth to move to `test-fixtures` module and have it only a single time?
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamRepartitionIntegrationTest.java:
##########
@@ -398,9 +405,18 @@ public void shouldUseStreamPartitionerForRepartitionOperation(final String topol
final Repartitioned<Integer, String> repartitioned = Repartitioned
.<Integer, String>as(repartitionName)
- .withStreamPartitioner((topic, key, value, numPartitions) -> {
- partitionerInvocation.incrementAndGet();
- return Optional.of(Collections.singleton(partition));
+ .withStreamPartitioner(new StreamPartitioner<>() {
+ @SuppressWarnings({"removal"})
Review Comment:
```suggestion
@SuppressWarnings("removal")
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java:
##########
@@ -1371,4 +1366,23 @@ public long extract(final ConsumerRecord<Object, Object> record, final long part
return DEFAULT_TIMESTAMP;
}
}
+
+ private static class FixedPartitionPartitioner implements StreamPartitioner<Integer, Object> {
Review Comment:
The removed code also uses `<String, String>`
```
private StreamPartitioner<String, String> constantPartitioner(final Integer partition)
```
--
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] MINOR: Suppress header-less partitioner deprecation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #23134:
URL: https://github.com/apache/kafka/pull/23134#issuecomment-5284126258
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] MINOR: Suppress header-less partitioner deprecation [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #23134:
URL: https://github.com/apache/kafka/pull/23134#discussion_r3764512669
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java:
##########
@@ -222,4 +224,23 @@ private Properties streamsConfiguration(final String safeTestName) {
config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
return config;
}
+
+ private static class FixedPartitionPartitioner implements StreamPartitioner<Integer, Object> {
Review Comment:
I mentioned it. Just wandering if 4 occurrences is enough to extract it
Don't think we expect any more uses of this partitoner + after 5.0 we're going to remove it (and we could back to lambda syntax)
--
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] MINOR: Suppress header-less partitioner deprecation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23134:
URL: https://github.com/apache/kafka/pull/23134#discussion_r3768022026
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java:
##########
@@ -222,4 +224,23 @@ private Properties streamsConfiguration(final String safeTestName) {
config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
return config;
}
+
+ private static class FixedPartitionPartitioner implements StreamPartitioner<Integer, Object> {
Review Comment:
Guess should be fine either way. (Don't think there is a "count threshold"; well, for production code there is, and it's 2 -- we should not have any duplication there. But for test-only code, we can handle it more loosley)
--
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] MINOR: Suppress header-less partitioner deprecation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23134:
URL: https://github.com/apache/kafka/pull/23134#discussion_r3770105834
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamRepartitionIntegrationTest.java:
##########
@@ -398,9 +405,18 @@ public void shouldUseStreamPartitionerForRepartitionOperation(final String topol
final Repartitioned<Integer, String> repartitioned = Repartitioned
.<Integer, String>as(repartitionName)
- .withStreamPartitioner((topic, key, value, numPartitions) -> {
- partitionerInvocation.incrementAndGet();
- return Optional.of(Collections.singleton(partition));
+ .withStreamPartitioner(new StreamPartitioner<>() {
+ @SuppressWarnings({"removal"})
Review Comment:
```suggestion
@SuppressWarnings("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] MINOR: Suppress header-less partitioner deprecation [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #23134:
URL: https://github.com/apache/kafka/pull/23134#discussion_r3769241436
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java:
##########
@@ -222,4 +224,23 @@ private Properties streamsConfiguration(final String safeTestName) {
config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
return config;
}
+
+ private static class FixedPartitionPartitioner implements StreamPartitioner<Integer, Object> {
Review Comment:
@mjsax
Thanks for comments
- Moved `FixedPartitionPartitioner` to `test-fixtures` (let me know if you disagree with package)
- Fixed tests, and re-run tests + int tests. All green localy
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java:
##########
@@ -222,4 +224,23 @@ private Properties streamsConfiguration(final String safeTestName) {
config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
return config;
}
+
+ private static class FixedPartitionPartitioner implements StreamPartitioner<Integer, Object> {
Review Comment:
@mjsax
Thanks for comments
- Moved `FixedPartitionPartitioner` to `test-fixtures` (let me know if you disagree with package)
- Fixed tests, and re-run tests + int tests. All green locally
--
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] MINOR: Suppress header-less partitioner deprecation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23134:
URL: https://github.com/apache/kafka/pull/23134#discussion_r3768022026
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java:
##########
@@ -222,4 +224,23 @@ private Properties streamsConfiguration(final String safeTestName) {
config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
return config;
}
+
+ private static class FixedPartitionPartitioner implements StreamPartitioner<Integer, Object> {
Review Comment:
Guess should be fine either way.
--
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] MINOR: Suppress header-less partitioner deprecation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #23134:
URL: https://github.com/apache/kafka/pull/23134
--
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