kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gwens...@apache.org
Subject [07/50] [abbrv] kafka git commit: KAFKA-3619: File handles are leaked on .lock files of ProcessorStateManager
Date Fri, 29 Apr 2016 22:05:21 GMT
KAFKA-3619: File handles are leaked on .lock files of ProcessorStateManager

Kafka Streams seems to hold file handles on the `.lock` files for the state dirs, resulting
in an explosion of filehandles over time. Running `lsof` shows the number of open filehandles
on the `.lock` file increasing rapidly over time. In a separate test project, I reproduced
the issue and determined that in order for the filehandle to be relinquished the `FileChannel`
instance must be properly closed. Applying this patch seems to resolve the issue in my job.

Author: Greg Fodor <gfodor@gmail.com>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #1267 from gfodor/bug/state-lock-filehandle-leak


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/996e29cf
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/996e29cf
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/996e29cf

Branch: refs/heads/0.10.0
Commit: 996e29cfe8a9e5a45d4b778a84fb20479eeba469
Parents: 1b764c5
Author: Greg Fodor <gfodor@gmail.com>
Authored: Mon Apr 25 13:45:51 2016 -0700
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Mon Apr 25 13:45:51 2016 -0700

----------------------------------------------------------------------
 .../streams/processor/internals/ProcessorStateManager.java      | 4 ++++
 .../apache/kafka/streams/processor/internals/StreamThread.java  | 1 +
 .../streams/processor/internals/ProcessorStateManagerTest.java  | 5 ++++-
 3 files changed, 9 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/996e29cf/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
index 003b988..0cdf44c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
@@ -134,6 +134,9 @@ public class ProcessorStateManager {
             retry--;
             lock = lockStateDirectory(channel);
         }
+        if (lock == null) {
+            channel.close();
+        }
         return lock;
     }
 
@@ -368,6 +371,7 @@ public class ProcessorStateManager {
         } finally {
             // release the state directory directoryLock
             directoryLock.release();
+            directoryLock.channel().close();
         }
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/996e29cf/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index f02683e..eff90e8 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -519,6 +519,7 @@ public class StreamThread extends Thread {
                                 if (directoryLock != null) {
                                     try {
                                         directoryLock.release();
+                                        directoryLock.channel().close();
                                     } catch (IOException e) {
                                         log.error("Failed to release the state directory
lock");
                                     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/996e29cf/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
index 84b59e6..e3669e8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
@@ -213,7 +213,10 @@ public class ProcessorStateManagerTest {
             try {
                 assertNotNull(lock);
             } finally {
-                if (lock != null) lock.release();
+                if (lock != null) {
+                    lock.release();
+                    lock.channel().close();
+                }
             }
         } finally {
             Utils.delete(baseDir);


Mime
View raw message