You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by "Bharath Kumarasubramanian (JIRA)" <ji...@apache.org> on 2019/07/22 20:44:00 UTC
[jira] [Closed] (SAMZA-1658) samza-kafka module use kafka
deprecated method to get topic metadata
[ https://issues.apache.org/jira/browse/SAMZA-1658?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Bharath Kumarasubramanian closed SAMZA-1658.
--------------------------------------------
> samza-kafka module use kafka deprecated method to get topic metadata
> --------------------------------------------------------------------
>
> Key: SAMZA-1658
> URL: https://issues.apache.org/jira/browse/SAMZA-1658
> Project: Samza
> Issue Type: Improvement
> Components: kafka
> Affects Versions: 0.12.0, 0.13.0, 0.14.0, 0.13.1
> Reporter: sailingYang
> Priority: Major
> Fix For: 0.15.0
>
>
> in ClientUtilTopicMetadataStore, samza-kafka use ClientUtils.fetchTopicMetadata to get topic metadata, this is a deprecated method in kafka. this method use SyncProducer and may get some problem.
> {code:java}
> // code placeholder
> def getTopicInfo(topics: Set[String]) = {
> val currCorrId = corrID.getAndIncrement
> val response: TopicMetadataResponse = ClientUtils.fetchTopicMetadata(topics, brokers, clientId, timeout, currCorrId)
> if (response.correlationId != currCorrId) {
> throw new SamzaException("CorrelationID did not match for request on topics %s (sent %d, got %d)" format (topics, currCorrId, response.correlationId))
> }
> response.topicsMetadata
> .map(metadata => (metadata.topic, metadata))
> .toMap
> }
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.14#76016)