flume-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From br...@apache.org
Subject [2/2] git commit: FLUME-1512: File Channel should not stop during a checkpoint
Date Fri, 24 Aug 2012 14:35:41 GMT
FLUME-1512: File Channel should not stop during a checkpoint

(Hari Shreedharan via Brock Noland)


Project: http://git-wip-us.apache.org/repos/asf/flume/repo
Commit: http://git-wip-us.apache.org/repos/asf/flume/commit/0a9924bb
Tree: http://git-wip-us.apache.org/repos/asf/flume/tree/0a9924bb
Diff: http://git-wip-us.apache.org/repos/asf/flume/diff/0a9924bb

Branch: refs/heads/trunk
Commit: 0a9924bb4eb6af86a247ff671bc3f219654c655f
Parents: 208f4e9
Author: Brock Noland <brock@apache.org>
Authored: Fri Aug 24 09:35:00 2012 -0500
Committer: Brock Noland <brock@apache.org>
Committed: Fri Aug 24 09:35:00 2012 -0500

----------------------------------------------------------------------
 .../java/org/apache/flume/channel/file/Log.java    |   56 ++++++++------
 1 files changed, 32 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flume/blob/0a9924bb/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
index 1e2706b..5b39b57 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
@@ -488,41 +488,49 @@ class Log {
     checkpointReadLock.unlock();
   }
 
+  private void lockExclusive(){
+    checkpointWriterLock.lock();
+  }
 
   /**
    * Synchronization required since we do not want this
    * to be called during a checkpoint.
    */
   synchronized void close() {
-    open = false;
-    if (worker != null) {
-      worker.shutdown();
-      worker.interrupt();
-    }
-    if (logFiles != null) {
-      for (int index = 0; index < logFiles.length(); index++) {
-        logFiles.get(index).close();
+    lockExclusive();
+    try {
+      open = false;
+      if (worker != null) {
+        worker.shutdown();
+        worker.interrupt();
       }
-    }
-    synchronized (idLogFileMap) {
-      for(Integer logId : idLogFileMap.keySet()) {
-        LogFile.RandomReader reader = idLogFileMap.get(logId);
-        if(reader != null) {
-          reader.close();
+      if (logFiles != null) {
+        for (int index = 0; index < logFiles.length(); index++) {
+          logFiles.get(index).close();
+        }
+      }
+      synchronized (idLogFileMap) {
+        for (Integer logId : idLogFileMap.keySet()) {
+          LogFile.RandomReader reader = idLogFileMap.get(logId);
+          if (reader != null) {
+            reader.close();
+          }
         }
       }
-    }
-    try {
-      unlock(checkpointDir);
-    } catch(IOException ex) {
-      LOGGER.warn("Error unlocking " + checkpointDir, ex);
-    }
-    for (File logDir : logDirs) {
       try {
-        unlock(logDir);
-      } catch(IOException ex) {
-        LOGGER.warn("Error unlocking " + logDir, ex);
+        unlock(checkpointDir);
+      } catch (IOException ex) {
+        LOGGER.warn("Error unlocking " + checkpointDir, ex);
       }
+      for (File logDir : logDirs) {
+        try {
+          unlock(logDir);
+        } catch (IOException ex) {
+          LOGGER.warn("Error unlocking " + logDir, ex);
+        }
+      }
+    } finally {
+      unlockExclusive();
     }
   }
 


Mime
View raw message