[
https://issues.apache.org/jira/browse/FLINK-6288?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15986124#comment-15986124
]
ASF GitHub Bot commented on FLINK-6288:
---------------------------------------
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/3766
Nice following the discussions here :) Let me wrap up the discussion so far:
The old way -
```
interface KafkaPartitioner {
void open(int[] partitions, int subtaskIndex, int numSubtasks);
int partition(T record, byte[] key, byte[] value, int numPartitions);
}
```
The (last) proposed new way -
```
interface FlinkKafkaPartitioner {
void open(int subtaskIndex, int numSubtask);
int partition(T record, byte[] key, byte[] value, String targetTopic,
int[] partitions)
}
```
and have an internal cache of partitioner informations: `Map<String,
PartitionerInfo>`.
The `PartitionerInfo` can actually just be the partition id array, I don't
think we need another wrapper class if we just need a single
`FlinkKafkaPartitioner` per subtask for all (including dynamic) topics.
I like the proposal of the new partitioner, as then users do not need to
provide multiple partitioners. Just the question with how well this works for
the general use case, because then implementations of the new `partition`
method need to handle different topics (which probably makes sense because we
want to generally treat topics as dynamic anyways ..). The new way can also
allow us to handle upscaled target topics in the future.
For migration, for the dummy wrapper delegation, I think we should just
mimc the wrong, old behaviour. That was the behaviour it had always been
anyways, so we should not try to alter the behaviour if the user is still using
the old API. Deprecation and Javadoc message is responsible of pushing them to
change to the new API.
> FlinkKafkaProducer's custom Partitioner is always invoked with number of
> partitions of default topic
> ----------------------------------------------------------------------------------------------------
>
> Key: FLINK-6288
> URL: https://issues.apache.org/jira/browse/FLINK-6288
> Project: Flink
> Issue Type: Improvement
> Components: Kafka Connector
> Reporter: Tzu-Li (Gordon) Tai
> Assignee: Fang Yong
>
> The {{FlinkKafkaProducerBase}} supports routing records to topics besides the
> default topic, but the custom {{Partitioner}} interface does not follow this
> semantic.
> The partitioner is always invoked the {{partition}} method with the number of
> partitions in the default topic, and not the number of partitions of the
> current {{targetTopic}}.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
