sailingYang created SAMZA-1658:

             Summary: samza-kafka module use kafka deprecated method to get 
topic metadata
                 Key: SAMZA-1658
             Project: Samza
          Issue Type: Improvement
          Components: kafka
    Affects Versions: 0.13.1, 0.14.0, 0.13.0, 0.12.0
            Reporter: sailingYang
             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 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))

    .map(metadata => (metadata.topic, metadata))

This message was sent by Atlassian JIRA

Reply via email to