Mailing-List: contact issues-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
Date: Thu, 27 Apr 2017 07:16:04 +0000 (UTC)
From: "ASF GitHub Bot (JIRA)" <jira@apache.org>
To: issues@flink.apache.org
Message-ID: <JIRA.13062896.1491831510000.63473.1493277364087@Atlassian.JIRA>
In-Reply-To: <JIRA.13062896.1491831510000@Atlassian.JIRA>
References: <JIRA.13062896.1491831510000@Atlassian.JIRA> <JIRA.13062896.1491831510683@jira-lw-us.apache.org>
Subject: [jira] [Commented] (FLINK-6288) FlinkKafkaProducer's custom
 Partitioner is always invoked with number of partitions of default topic
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Thu, 27 Apr 2017 07:16:08 -0000


    [ https://issues.apache.org/jira/browse/FLINK-6288?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15986135#comment-15986135 ] 

ASF GitHub Bot commented on FLINK-6288:
---------------------------------------

Github user tzulitai commented on the issue:

    https://github.com/apache/flink/pull/3766
  
    One thing to be careful with, though:
    Since now we're querying Kafka for partition metadata within the `invoke` method, the query must be handled robustly and make sure it doesn't result in unexpectedly longer checkpoint times by blocking the whole stream at the Kafka sink.
    
    Most notably, we need to consider the corner case where Kafka isn't cooperating nicely:
    1. how to handle arbitrary long response time for fetching the partition metadata?
    2. how to handle the case where, due to some Kafka brokers temporary unavailable, the returned partitions is not complete?
    
    For 2., I can also forsee that we have a separate "partitions update thread" that refreshes the `Map<String, int[]>` cache continuously at a fixed interval. This can also involve to a `FlinkKafkaPartitioner` that can provide dynamically changing `int[] partitions` when invoking the `partition` method.
    
    Perhaps we shouldn't include that with this PR, as its orthogonal to the API change. Just some food for though :)


> 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)