KafkaAvroDeserializer Issue
<p>I am trying to desrialize a Kafka Avro message using Avro Kafka Deserializer. This code is very common and already in practice by lot of users. But I am facing few difficulties in implementing the same:
Code:</p>

<pre><code>Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("group.id", "CountryCounter");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("schema.registry.url", schemaUrl);
String topic = "customerContacts"

KafkaConsumer consumer = new KafkaConsumer(createConsumerConfig(brokers, groupId, url));
consumer.subscribe(Collections.singletonList(topic));

System.out.println("Reading topic:" + topic);

while (true) {
        @SuppressWarnings("unchecked")
        ConsumerRecords&lt;String, ATPEvent&gt; records = (ConsumerRecords&lt;String, ATPEvent&gt;) consumer.poll(1000);  //1

       for (ConsumerRecord&lt;String, ATPEvent&gt; record: records)  //2 {

            try {
                kafkaMessageInputStream = new ByteBufferInputStream(Lists.newArrayList(ByteBuffer.wrap(record.value()))); //3
                avroBinaryDecoder = avroDecoderFactory.binaryDecoder(kafkaMessageInputStream, avroBinaryDecoder);
                avroEvent = reader.read(avroEvent, avroBinaryDecoder);
                System.out.println(avroEvent);
                kafkaMessageInputStream.close();
              } catch (Exception ex) {
                System.out.println("Unable to process event from kafka, see exception details" + ex);
              }
        }
        consumer.commitSync(); //4
}
</code></pre>

<p>Now, here are the 4 issue:</p>

<ol>
<li><p>I have to add cast, otherwise it will through an error as 
<code>Type mismatch: cannot convert from Map&lt;String,ConsumerRecords&lt;String,ATPEvent&gt;&gt; to ConsumerRecords&lt;String,ATPEvent&gt;</code></p></li>
<li><p><code>Can only iterate over an array or an instance of java.lang.Iterable</code> 
I have no clue why this? Can I do this? : </p>

<pre><code>List&lt;ConsumerRecord&lt;String, ATPEvent&gt;&gt; records = (List&lt;ConsumerRecord&lt;String, ATPEvent&gt;&gt;) consumer.poll(1000);

for (ConsumerRecord&lt;String, ATPEvent&gt; record: records) {
</code></pre></li>
<li><p><code>The method wrap(byte[]) in the type ByteBuffer is not applicable for the arguments (ATPEvent)</code> 
This I understand, but how can I convert a class to a byte, is there any other way?</p></li>
<li><p><code>The method commitSync() is undefined for the type KafkaConsumer&lt;String,ATPEvent&gt;</code> 
Can I just use <code>consumer.close();</code></p></li>
</ol>

<p>Please provide solution for 2 &amp; 3, and if possible explanation for 1 &amp; 4.</p>