卡夫卡的高級消費者API生命週期0.8

Question

我無法找到高級消費者的生命週期描述。我在0.8.2.2上，我不能使用來自kafka客戶的「現代」消費者。這裏是我的代碼：

def consume(numberOfEvents: Int, await: Duration = 100.millis): List[MessageEnvelope] = { 
    val consumerProperties = new Properties() 
    consumerProperties.put("zookeeper.connect", kafkaConfig.zooKeeperConnectString) 
    consumerProperties.put("group.id", consumerGroup) 
    consumerProperties.put("auto.offset.reset", "smallest") 

    val consumer = Consumer.create(new ConsumerConfig(consumerProperties)) 

    try { 
     val messageStreams = consumer.createMessageStreams(
     Predef.Map(kafkaConfig.topic -> 1), 
     new DefaultDecoder, 
     new MessageEnvelopeDecoder) 

     val receiveMessageFuture = Future[List[MessageEnvelope]] { 
     messageStreams(kafkaConfig.topic) 
      .flatMap(stream => stream.take(numberOfEvents).map(_.message())) 
     } 

     Await.result(receiveMessageFuture, await) 
    } finally { 
     consumer.shutdown() 
    } 

我不清楚。我應該在每次檢索消息後關閉消費者，還是保留實例並將其重新用於消息提取？我認爲重用實例是正確的方式，但找不到一些文章/最佳實踐。

我想重用消費者和/或messageStreams。它不適合我，我找不到原因。

如果我嘗試重用messageStreams，我得到異常：

2017-04-17_19:57:57.088 ERROR MessageEnvelopeConsumer - Error while awaiting for messages java.lang.IllegalStateException: Iterator is in failed state 
java.lang.IllegalStateException: Iterator is in failed state 
    at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:54) 
    at scala.collection.IterableLike$class.take(IterableLike.scala:134) 
    at kafka.consumer.KafkaStream.take(KafkaStream.scala:25) 

發生在這裏：

def consume(numberOfEvents: Int, await: Duration = 100.millis): List[MessageEnvelope] = { 
    try { 
     val receiveMessageFuture = Future[List[MessageEnvelope]] { 
     messageStreams(kafkaConfig.topic) 
      .flatMap(stream => stream.take(numberOfEvents).map(_.message())) 
     } 
     Try(Await.result(receiveMessageFuture, await)) match { 
     case Success(result) => result 
     case Failure(_: TimeoutException) => List.empty 
     case Failure(e) => 
      // ===> never got any message from topic 
      logger.error(s"Error while awaiting for messages ${e.getClass.getName}: ${e.getMessage}", e) 
      List.empty 

     } 
    } catch { 
     case e: Exception => 
     logger.warn(s"Error while consuming messages", e) 
     List.empty 
    } 
    } 

我試圖創建messageStreams每次：

沒有運氣...

2017-04-17_20:02:44.236 WARN MessageEnvelopeConsumer - Error while consuming messages 
kafka.common.MessageStreamsExistException: ZookeeperConsumerConnector can create message streams at most once 
    at kafka.consumer.ZookeeperConsumerConnector.createMessageStreams(ZookeeperConsumerConnector.scala:151) 
    at MessageEnvelopeConsumer.consume(MessageEnvelopeConsumer.scala:47) 

Ha ppens這裏：

def consume(numberOfEvents: Int, await: Duration = 100.millis): List[MessageEnvelope] = { 
    try { 

     val messageStreams = consumer.createMessageStreams(
     Predef.Map(kafkaConfig.topic -> 1), 
     new DefaultDecoder, 
     new MessageEnvelopeDecoder) 

     val receiveMessageFuture = Future[List[MessageEnvelope]] { 
     messageStreams(kafkaConfig.topic) 
      .flatMap(stream => stream.take(numberOfEvents).map(_.message())) 
     } 
     Try(Await.result(receiveMessageFuture, await)) match { 
     case Success(result) => result 
     case Failure(_: TimeoutException) => List.empty 
     case Failure(e) => 
      logger.error(s"Error while awaiting for messages ${e.getClass.getName}: ${e.getMessage}", e) 
      List.empty 

     } 
    } catch { 
     case e: Exception => 
     // ===> now exception raised here 
     logger.warn(s"Error while consuming messages", e) 
     List.empty 
    } 
    } 

UPD

我使用的基於迭代器的方法。它看起來是這樣的：

// consumerProperties.put("consumer.timeout.ms", "100")  

private lazy val consumer: ConsumerConnector = Consumer.create(new ConsumerConfig(consumerProperties)) 

    private lazy val messageStreams: Seq[KafkaStream[Array[Byte], MessageEnvelope]] = 
    consumer.createMessageStreamsByFilter(Whitelist(kafkaConfig.topic), 1, new DefaultDecoder, new MessageEnvelopeDecoder) 


    private lazy val iterator: ConsumerIterator[Array[Byte], MessageEnvelope] = { 
    val stream = messageStreams.head 
    stream.iterator() 
    } 

    def consume(): List[MessageEnvelope] = { 
    try { 
     if (iterator.hasNext) { 
     val fromKafka: MessageAndMetadata[Array[Byte], MessageEnvelope] = iterator.next 
     List(fromKafka.message()) 
     } else { 
     List.empty 
     } 

    } catch { 
     case _: ConsumerTimeoutException => 
     List.empty 

     case e: Exception => 
     logger.warn(s"Error while consuming messages", e) 
     List.empty 
    } 
    } 

現在我想弄清楚它是否會自動提交偏移ZK ...

Answer 1

我的答案是最新的問題更新。迭代器方法可以像預期的那樣爲我工作。

Answer 2

恆關機導致不必要的消費羣體重新平衡從而影響性能提升不少。請參閱本文以獲得最佳做法：https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example