APEXMALHAR-2497 APEXMALHAR-2162 1) Refactor the Exactly Once output operator. 2) Refactor and fix the issues of unit tests.

Author: chaithu14

examples/sql/pom.xml

@@ -90,6 +90,19 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>org.apache.apex</groupId>
+      <artifactId>malhar-kafka-common</artifactId>
+      <version>${project.parent.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>*</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka_2.11</artifactId>

kafka/kafka-common/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaConsumer.java

@@ -27,6 +27,7 @@
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 
 /**
@@ -53,7 +54,7 @@ public interface AbstractKafkaConsumer
    * @param timeOut time in milliseconds, spent waiting in poll if data is not available in buffer.
    * @return records
    */
-  ConsumerRecords<byte[], byte[]> pollRecords(long timeOut);
+  ConsumerRecords pollRecords(long timeOut);
 
   /**
    * Commit the specified offsets for the specified list of topics and partitions to Kafka.
@@ -124,4 +125,6 @@ public interface AbstractKafkaConsumer
    * @param tp partition
    */
   long positionPartition(TopicPartition tp);
+
+  List<PartitionInfo> partitionsFor(String topic);
 }