Kafka 0.9.0.1 for JDK 7

wojtek-szymanski · Mar 20, 2016 · 74129bd · 74129bd
1 parent c44506b
commit 74129bd
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 13 deletions.
diff --git a/pom.xml b/pom.xml
@@ -20,7 +20,7 @@
 
     <groupId>com.github.charithe</groupId>
     <artifactId>kafka-junit</artifactId>
-    <version>2.1-SNAPSHOT</version>
+    <version>1.9.1-SNAPSHOT</version>
 
     <name>kafka-junit</name>
     <description>JUnit rule to spin-up a Kafka broker during tests</description>
@@ -42,7 +42,7 @@
 
 
     <properties>
-        <java.version>1.8</java.version>
+        <java.version>1.7</java.version>
         <kafka.version>0.9.0.1</kafka.version>
     </properties>
 

diff --git a/src/main/java/com/github/charithe/kafka/KafkaJunitRule.java b/src/main/java/com/github/charithe/kafka/KafkaJunitRule.java
@@ -44,6 +44,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -272,17 +273,21 @@ public <T> List<T> readMessages(final KafkaConsumer<T, T> consumer, final String
         ExecutorService singleThread = Executors.newSingleThreadExecutor();
         try {
             consumer.subscribe(Lists.newArrayList(topic));
-            Future<List<T>> future = singleThread.submit(() -> {
-                                                             List<T> messages = new ArrayList<>(expectedMessages);
-                                                             while (messages.size() < expectedMessages) {
-                                                                 ConsumerRecords<T, T> records = consumer.poll(POLL_TIMEOUT_MS);
-                                                                 for (ConsumerRecord<T, T> rec : records) {
-                                                                     LOGGER.debug("Received message: {} -> {}", rec.key(), rec.value());
-                                                                     messages.add(rec.value());
-                                                                 }
-                                                             }
-                                                             return messages;
-                                                         }
+            Future<List<T>> future = singleThread.submit(
+                    new Callable<List<T>>() {
+                        @Override
+                        public List<T> call() throws Exception {
+                            List<T> messages = new ArrayList<>(expectedMessages);
+                            while (messages.size() < expectedMessages) {
+                                ConsumerRecords<T, T> records = consumer.poll(POLL_TIMEOUT_MS);
+                                for (ConsumerRecord<T, T> rec : records) {
+                                    LOGGER.debug("Received message: {} -> {}", rec.key(), rec.value());
+                                    messages.add(rec.value());
+                                }
+                            }
+                            return messages;
+                        }
+                    }
             );
 
             return future.get(timeoutSeconds, SECONDS);