Index: streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java	(date 1535224105000)
+++ streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java	(revision )
@@ -216,12 +216,14 @@
         final LockAndOwner lockAndOwner = locks.get(taskId);
         if (lockAndOwner != null && lockAndOwner.owningThread.equals(Thread.currentThread().getName())) {
             locks.remove(taskId);
-            lockAndOwner.lock.release();
-            log.debug("{} Released state dir lock for task {}", logPrefix(), taskId);
-
-            final FileChannel fileChannel = channels.remove(taskId);
-            if (fileChannel != null) {
-                fileChannel.close();
+            try {
+                lockAndOwner.lock.release();
+                log.debug("{} Released state dir lock for task {}", logPrefix(), taskId);
+            } finally {
+                final FileChannel fileChannel = channels.remove(taskId);
+                if (fileChannel != null) {
+                    fileChannel.close();
+                }
             }
         }
     }