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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
