[SAMZA-1658] samza-kafka module use kafka deprecated method to get topic metadata

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.12.0, 0.13.0, 0.14.0, 0.13.1
• Fix Version/s: 0.15.0
• Component/s: kafka
• Assignee: Unassigned
• Reporter: sailingYang
• Created: 15/Apr/18 07:56
• Updated: 22/Jul/19 20:43
• Resolved: 22/Jul/19 20:43

Description

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

  response.topicsMetadata
    .map(metadata => (metadata.topic, metadata))
    .toMap
}