APEXMALHAR-2497 APEXMALHAR-2162 1) Refactor the Exactly Once output o…

…perator. 2) Refactor and fix the issues of unit tests.
apache · Nov 10, 2017 · 15d2903 · 15d2903
1 parent 9090578
commit 15d2903
Show file tree

Hide file tree

Showing 26 changed files with 1,821 additions and 2,337 deletions.
diff --git a/examples/sql/pom.xml b/examples/sql/pom.xml
@@ -84,6 +84,19 @@
       <type>test-jar</type>
       <scope>test</scope>
     </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>

diff --git a/kafka/kafka-common/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaConsumer.java b/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);
 }