title: Upgraded to version 0.10 of "New Apache Kafka Clients Consumer" from version 0.9 "Apache Kafka Clients Consumer"
body: <p>My current product is Apache Kafka Clients Consumer 0.9. I plan to upgrade to New Apache Kafka Clients Consumer 0.10.1.0. I have already completed the development of the API change and the product is consuming the message from the Kafka broker (v.0.10), so I do not know how to replace it.</p>
  
  <p>What I worry about is whether the offset will be kept safe by simply replacing the upgraded product with New Kafka Consumer 0.10.1. Of course, existing products will have to work off, and the consumer group will be the same.</p>
  
  <p>The API changes are as follows.</p>
  
  <pre><code>// OLD KAFKA CONSUMER API
  val consumerConnector = Consumer.create(consumerProperties)
  val kafkaStreams: List[KafkaStream[Array[Byte], Array[Byte]]] = consumerConnector.createMessageStreams(threadCount).get(topic).get
  
  kafkaStreams map (stream =&gt; {
      val iterator: ConsumerIterator[Array[Byte], Array[Byte]] = stream.iterator
  
      while (true) {
          // process iterator.next.message
      }
  })
  
  // NEW KAFKA CONSUMER API
  val kafkaConsumerList = List[KafkaConsumer[Array[Byte], Array[Byte]]
  
  for (i &lt;- 1 to threadCount) {
      val kafkaConsumer: KafkaConsumer[Array[Byte], Array[Byte]] = new KafkaConsumer[Array[Byte], Array[Byte]](consumerProperties)
      kafkaConsumerList = kafkaConsumer :: kafkaConsumerList
  }
  
  kafkaConsumerList map (consumer =&gt; {
      consumer.subscribe(topics)
  
      while (true) {
          val records: ConsumerRecords[Array[Byte], Array[Byte]] = consumer.poll(pollTimeout)
  
          while (records.iterator.hasNext) {
              // process records.iterator.next.value
          }
      }
  })
  </code></pre>
  
  <p>For reference, I am using the auto offset method, and I will continue to do so.</p>