From 8bd329d3f8edfb0ad817da2cc9c205ef4f647c94 Mon Sep 17 00:00:00 2001
From: Jeremy Custenborder <jeremy@scarcemedia.com>
Date: Fri, 12 Feb 2016 13:46:30 -0800
Subject: [PATCH] KAFKA-3225 Added test case for missing commit call on
 WorkerSourceTask. Added commitSourceTask() to follow finishSuccessfulFlush()
 which calls SourceTask.commit() and log exceptions.

---
 .../kafka/connect/runtime/WorkerSourceTask.java   | 15 +++++++++++++++
 .../connect/runtime/WorkerSourceTaskTest.java     |  2 ++
 2 files changed, 17 insertions(+)

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
index 30c22620f14ad..562e03e5f0786 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
@@ -290,6 +290,8 @@ public boolean commitOffsets() {
                 finishSuccessfulFlush();
                 log.debug("Finished {} offset commitOffsets successfully in {} ms",
                         this, time.milliseconds() - started);
+
+                commitSourceTask();
                 return true;
             }
         }
@@ -334,9 +336,22 @@ public void onCompletion(Throwable error, Void result) {
         finishSuccessfulFlush();
         log.info("Finished {} commitOffsets successfully in {} ms",
                 this, time.milliseconds() - started);
+
+        commitSourceTask();
+
         return true;
     }
 
+    private void commitSourceTask() {
+        try {
+            this.task.commit();
+        } catch (InterruptedException ex) {
+            log.warn("Commit interrupted", ex);
+        } catch (Throwable ex) {
+            log.error("Exception thrown while calling task.commit()", ex);
+        }
+    }
+
     private synchronized void finishFailedFlush() {
         offsetWriter.cancelFlush();
         outstandingMessages.putAll(outstandingMessagesBacklog);
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
index 1f557e45e4075..59e9b862455d3 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
@@ -158,6 +158,8 @@ public void testCommit() throws Exception {
         final CountDownLatch pollLatch = expectPolls(1);
         expectOffsetFlush(true);
 
+        sourceTask.commit();
+        EasyMock.expectLastCall();
         sourceTask.stop();
         EasyMock.expectLastCall();
         expectOffsetFlush(true);