From b914c2bccf914f154b043dbbdf17df56ed00ccdb Mon Sep 17 00:00:00 2001
From: Jonathan Hsieh <jon@cloudera.com>
Date: Wed, 15 Jun 2011 00:43:53 -0700
Subject: [PATCH] FLUME-658: Handle Unexpected RuntimeExceptions diffrerently than IOExceptions in sinks and decorators

---
 .../handlers/debug/InsistentAppendDecorator.java   |    8 ++-
 .../handlers/debug/InsistentOpenDecorator.java     |   14 ++-
 .../flume/handlers/debug/StubbornAppendSink.java   |    8 ++-
 .../flume/collector/TestCollectorSink.java         |  105 ++++++++++++++++++++
 4 files changed, 130 insertions(+), 5 deletions(-)

diff --git a/flume-core/src/main/java/com/cloudera/flume/handlers/debug/InsistentAppendDecorator.java b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/InsistentAppendDecorator.java
index 755428c..9f2365a 100644
--- a/flume-core/src/main/java/com/cloudera/flume/handlers/debug/InsistentAppendDecorator.java
+++ b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/InsistentAppendDecorator.java
@@ -113,7 +113,7 @@ public class InsistentAppendDecorator<S extends EventSink> extends
         return;
       } catch (InterruptedException ie) {
         throw ie;
-      } catch (Exception e) {
+      } catch (IOException e) {
         // this is an unexpected exception
         long waitTime = backoff.sleepIncrement();
         LOG.info("append attempt " + attemptRetries + " failed, backoff ("
@@ -131,6 +131,12 @@ public class InsistentAppendDecorator<S extends EventSink> extends
           attemptRetries++;
           appendRetries++;
         }
+      } catch (RuntimeException e) {
+        // this is an unexpected exception
+        LOG.info("Failed due to unexpected runtime exception "
+            + "during append attempt", e);
+        appendGiveups++;
+        throw e;
       }
     }
     appendGiveups++;
diff --git a/flume-core/src/main/java/com/cloudera/flume/handlers/debug/InsistentOpenDecorator.java b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/InsistentOpenDecorator.java
index b96a76f..b23ddf6 100644
--- a/flume-core/src/main/java/com/cloudera/flume/handlers/debug/InsistentOpenDecorator.java
+++ b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/InsistentOpenDecorator.java
@@ -119,7 +119,7 @@ public class InsistentOpenDecorator<S extends EventSink> extends
       } catch (InterruptedException ie) {
         // Not open so no need to close
         throw ie;
-      } catch (Exception e) {
+      } catch (IOException e) {
 
         if (!opening) {
           throw new IOException("Unable to open and then close requested");
@@ -137,11 +137,19 @@ public class InsistentOpenDecorator<S extends EventSink> extends
           backoff.waitUntilRetryOk();
         } catch (InterruptedException e1) {
           // got an interrupted signal, bail out!
+          opening = false;
+          openGiveups++;
           throw e1;
         }
 
         attemptRetries++;
         openRetries++;
+      } catch (RuntimeException rte) {
+        LOG.error("Failed due to unexpected runtime exception in open retryer",
+            rte);
+        openGiveups++;
+        opening = false;
+        throw rte;
       }
     }
     openGiveups++;
@@ -234,8 +242,8 @@ public class InsistentOpenDecorator<S extends EventSink> extends
     ReportEvent rpt = super.getReport();
 
     // parameters
-    rpt.hierarchicalMerge("backoffPolicy." + backoff.getName(), backoff
-        .getMetrics());
+    rpt.hierarchicalMerge("backoffPolicy." + backoff.getName(),
+        backoff.getMetrics());
 
     // counters
     rpt.setLongMetric(A_REQUESTS, openRequests);
diff --git a/flume-core/src/main/java/com/cloudera/flume/handlers/debug/StubbornAppendSink.java b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/StubbornAppendSink.java
index 1609ac5..9436eb4 100644
--- a/flume-core/src/main/java/com/cloudera/flume/handlers/debug/StubbornAppendSink.java
+++ b/flume-core/src/main/java/com/cloudera/flume/handlers/debug/StubbornAppendSink.java
@@ -78,7 +78,7 @@ public class StubbornAppendSink<S extends EventSink> extends
       super.close(); // close
       throw ie;
 
-    } catch (Exception ex) {
+    } catch (IOException ex) {
       LOG.info("append failed on event '{}' with error: {}", e, ex.getMessage());
 
       appendFails.incrementAndGet();
@@ -89,6 +89,12 @@ public class StubbornAppendSink<S extends EventSink> extends
       appendSuccesses.incrementAndGet();
       // another exception may have been thrown at close/open/append
       appendRecovers.incrementAndGet();
+    } catch (RuntimeException ex) {
+      LOG.info("append failed on event '{}' with error: {}", e, ex.getMessage());
+
+      appendFails.incrementAndGet();
+      super.close(); // close
+      throw ex;
     }
   }
 
diff --git a/flume-core/src/test/java/com/cloudera/flume/collector/TestCollectorSink.java b/flume-core/src/test/java/com/cloudera/flume/collector/TestCollectorSink.java
index 48cf495..1f13c7f 100644
--- a/flume-core/src/test/java/com/cloudera/flume/collector/TestCollectorSink.java
+++ b/flume-core/src/test/java/com/cloudera/flume/collector/TestCollectorSink.java
@@ -20,11 +20,15 @@ package com.cloudera.flume.collector;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.Date;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -43,9 +47,12 @@ import com.cloudera.flume.agent.durability.WALManager;
 import com.cloudera.flume.conf.Context;
 import com.cloudera.flume.conf.FlumeArgException;
 import com.cloudera.flume.conf.FlumeBuilder;
+import com.cloudera.flume.conf.FlumeConfiguration;
 import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.conf.LogicalNodeContext;
 import com.cloudera.flume.conf.ReportTestingContext;
+import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
+import com.cloudera.flume.conf.SinkFactoryImpl;
 import com.cloudera.flume.core.Event;
 import com.cloudera.flume.core.EventImpl;
 import com.cloudera.flume.core.EventSink;
@@ -62,6 +69,8 @@ import com.cloudera.flume.handlers.hdfs.EscapedCustomDfsSink;
 import com.cloudera.flume.handlers.rolling.ProcessTagger;
 import com.cloudera.flume.handlers.rolling.RollSink;
 import com.cloudera.flume.handlers.rolling.Tagger;
+import com.cloudera.flume.handlers.thrift.Priority;
+import com.cloudera.flume.handlers.thrift.ThriftFlumeEvent;
 import com.cloudera.flume.reporter.ReportEvent;
 import com.cloudera.flume.reporter.ReportManager;
 import com.cloudera.util.BenchmarkHarness;
@@ -752,4 +761,100 @@ public class TestCollectorSink {
     ReportEvent rptb = ReportManager.get().getReportable("bar").getMetrics();
     assertEquals(1, (long) rptb.getLongMetric("bar"));
   }
+
+  /**
+   * This test verifies that a runtime exception (such as NPE, ArrayOutOfBounds,
+   * IllegalStateException)on an open of a subsink inside a collector does not
+   * hang a logical node
+   *
+   * @throws IOException
+   * @throws InterruptedException
+   * @throws FlumeSpecException
+   */
+  @Test(expected = RuntimeException.class)
+  public void testOpenRuntimeExceptionSink() throws IOException,
+      InterruptedException, FlumeSpecException {
+    final EventSink snk = mock(EventSink.class);
+    doThrow(new RuntimeException("Forced unexpected open error")).when(snk)
+        .open();
+    SinkBuilder sb = new SinkBuilder() {
+      @Override
+      public EventSink build(Context context, String... argv) {
+        return snk;
+      }
+    };
+    SinkFactoryImpl sf = new SinkFactoryImpl();
+    sf.setSink("cnf", sb);
+    FlumeBuilder.setSinkFactory(sf);
+
+    final EventSink coll = FlumeBuilder.buildSink(
+        LogicalNodeContext.testingContext(), "collector(5000) { cnf }");
+    coll.open();
+    coll.close();
+  }
+
+  /**
+   * This test verifies that a runtime exception (such as NPE, ArrayOutOfBounds,
+   * IllegalStateException) on an open of a subsink inside a collector does not
+   * hang a logical node
+   *
+   * @throws IOException
+   * @throws InterruptedException
+   * @throws FlumeSpecException
+   */
+  @Test(expected = RuntimeException.class)
+  public void testAppendRuntimeExceptionSink() throws IOException,
+      InterruptedException, FlumeSpecException {
+    final EventSink snk = mock(EventSink.class);
+    doThrow(new RuntimeException("Force unexpected append error")).when(snk)
+        .append((Event) anyObject());
+    SinkBuilder sb = new SinkBuilder() {
+      @Override
+      public EventSink build(Context context, String... argv) {
+        return snk;
+      }
+    };
+    SinkFactoryImpl sf = new SinkFactoryImpl();
+    sf.setSink("rte", sb);
+    FlumeBuilder.setSinkFactory(sf);
+
+    final EventSink coll = FlumeBuilder.buildSink(
+        LogicalNodeContext.testingContext(), "collector(5000) { rte }");
+    coll.open();
+    coll.append(new EventImpl("foo".getBytes()));
+    coll.close();
+  }
+
+  /**
+   * This test verifies that a runtime exception (such as NPE, ArrayOutOfBounds,
+   * IllegalStateException) on an close of a subsink inside a collector does not
+   * hang a logical node
+   *
+   * @throws IOException
+   * @throws InterruptedException
+   * @throws FlumeSpecException
+   */
+  @Test(expected = RuntimeException.class)
+  public void testCloseRuntimeExceptionSink() throws IOException,
+      InterruptedException, FlumeSpecException {
+    final EventSink snk = mock(EventSink.class);
+    doThrow(new RuntimeException("Force unexpected append error")).when(snk)
+        .close();
+    SinkBuilder sb = new SinkBuilder() {
+      @Override
+      public EventSink build(Context context, String... argv) {
+        return snk;
+      }
+    };
+    SinkFactoryImpl sf = new SinkFactoryImpl();
+    sf.setSink("closeRte", sb);
+    FlumeBuilder.setSinkFactory(sf);
+
+    final EventSink coll = FlumeBuilder.buildSink(
+        LogicalNodeContext.testingContext(), "collector(5000) { closeRte }");
+    coll.open();
+    coll.append(new EventImpl("foo".getBytes()));
+    coll.close();
+  }
+
 }
-- 
1.7.1

