svn commit: r1177806 - in /incubator/flume/trunk/flume-core/src: main/java/com/cloudera/flume/handlers/rolling/RollSink.java test/java/com/cloudera/flume/handlers/rolling/TestRollSink.java

[jm...@apache.org]

Author: jmhsieh
Date: Fri Sep 30 19:54:20 2011
New Revision: 1177806

URL: http://svn.apache.org/viewvc?rev=1177806&view=rev
Log:
FLUME-768: Agent deadlock possible due to blocked latch in driver thread (Prasad Mujumdar)


Modified:
    incubator/flume/trunk/flume-core/src/main/java/com/cloudera/flume/handlers/rolling/RollSink.java
    incubator/flume/trunk/flume-core/src/test/java/com/cloudera/flume/handlers/rolling/TestRollSink.java

Modified: incubator/flume/trunk/flume-core/src/main/java/com/cloudera/flume/handlers/rolling/RollSink.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-core/src/main/java/com/cloudera/flume/handlers/rolling/RollSink.java?rev=1177806&r1=1177805&r2=1177806&view=diff
==============================================================================
--- incubator/flume/trunk/flume-core/src/main/java/com/cloudera/flume/handlers/rolling/RollSink.java (original)
+++ incubator/flume/trunk/flume-core/src/main/java/com/cloudera/flume/handlers/rolling/RollSink.java Fri Sep 30 19:54:20 2011
@@ -148,15 +148,15 @@ public class RollSink extends EventSink.
             Clock.sleep(checkLatencyMs);
           } catch (InterruptedException e) {
             LOG.debug("TriggerThread interrupted");
-            doneLatch.countDown();
             return;
           }
         }
       } catch (InterruptedException e) {
         LOG.error("RollSink interrupted", e);
-      }
+      } finally {
       LOG.debug("TriggerThread shutdown");
       doneLatch.countDown();
+      }
     }
   };
 

Modified: incubator/flume/trunk/flume-core/src/test/java/com/cloudera/flume/handlers/rolling/TestRollSink.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-core/src/test/java/com/cloudera/flume/handlers/rolling/TestRollSink.java?rev=1177806&r1=1177805&r2=1177806&view=diff
==============================================================================
--- incubator/flume/trunk/flume-core/src/test/java/com/cloudera/flume/handlers/rolling/TestRollSink.java (original)
+++ incubator/flume/trunk/flume-core/src/test/java/com/cloudera/flume/handlers/rolling/TestRollSink.java Fri Sep 30 19:54:20 2011
@@ -389,4 +389,53 @@ public class TestRollSink {
     // See above for why there are 4 triggers:
     assertEquals(4, (long) rpt.getLongMetric(RollSink.A_ROLLS));
   }
+
+  /**
+   * This verifies that when roller's trigger aborts, the sink closes correctly
+   */
+  @Test(timeout=30000)
+  public void testTriggerAborted() throws IOException,
+      InterruptedException {
+
+    Tagger tagger = new ProcessTagger() {
+      @Override
+      public String getTag() {
+        return "-testtag";
+      }
+
+      @Override
+      public String newTag() {
+        // throw exception from the Trigger Thread
+        if (Thread.currentThread().getName().contains("Roll-TriggerThread"))
+          throw new RuntimeException("testExp");
+        return "-testtag";
+      }
+    };
+
+    final File f = FileUtil.mktempdir();
+    RollSink snk = new RollSink(new Context(), "test", new TimeTrigger(tagger,
+        10000), 250) {
+      @Override
+      protected EventSink newSink(Context ctx) throws IOException {
+        return new EscapedCustomDfsSink("file:///" + f.getPath(),
+            "sub-%{service}%{rolltag}");
+      }
+    };
+
+    Event e = new EventImpl("this is a test message".getBytes());
+    Attributes.setString(e, "service", "foo");
+    snk.open();
+    snk.append(e);
+    // wait for the trigger thread to abort down
+    while (snk.triggerThread.isAlive()) {
+      Clock.sleep(100);
+    }
+    snk.close();
+    // verify that the trigger thread did the cleanup
+    assertTrue(snk.triggerThread.doneLatch.getCount() == 0);
+    File fo = new File(f.getPath() + "/sub-foo-testtag");
+    assertTrue(fo.exists());
+    FileUtil.rmr(f);
+  }
+
 }