From 6ca03e90a476291cc312b52eb2546cd23482be1b Mon Sep 17 00:00:00 2001
From: Jonathan Hsieh <jon@cloudera.com>
Date: Fri, 7 Jan 2011 14:57:42 -0800
Subject: [PATCH] FLUME-462: cleanup type casting warnings in core build

---
 .../flume/agent/diskfailover/DiskFailoverDeco.java |   17 +++++-----
 .../flume/agent/durability/NaiveFileWALDeco.java   |   23 ++++++-------
 .../agent/TestEndToEndAckFailureRecovery.java      |   30 +++++++++---------
 .../agent/diskfailover/TestConcurrentDFOMan.java   |    9 +++--
 .../agent/diskfailover/TestDiskFailoverDeco.java   |   11 +++---
 .../agent/durability/TestConcurrentWALMan.java     |    4 +-
 .../agent/durability/TestNaiveFileWALDeco.java     |   34 +++++++++-----------
 .../TestNaiveFileWALManagerConcurrently.java       |   12 +++----
 8 files changed, 65 insertions(+), 75 deletions(-)

diff --git a/src/java/com/cloudera/flume/agent/diskfailover/DiskFailoverDeco.java b/src/java/com/cloudera/flume/agent/diskfailover/DiskFailoverDeco.java
index d1e43fd..f868ca3 100644
--- a/src/java/com/cloudera/flume/agent/diskfailover/DiskFailoverDeco.java
+++ b/src/java/com/cloudera/flume/agent/diskfailover/DiskFailoverDeco.java
@@ -48,8 +48,7 @@ import com.google.common.base.Preconditions;
  * It has a subordinate thread that drains the events that have been written to
  * disk. Latches are used to maintain open and close semantics.
  */
-public class DiskFailoverDeco<S extends EventSink> extends
-    EventSinkDecorator<S> {
+public class DiskFailoverDeco extends EventSinkDecorator<EventSink> {
   static final Logger LOG = LoggerFactory.getLogger(DiskFailoverDeco.class);
 
   final DiskFailoverManager dfoMan;
@@ -67,17 +66,17 @@ public class DiskFailoverDeco<S extends EventSink> extends
   final long checkmillis;
   final Context ctx;
 
-  public DiskFailoverDeco(S s, Context ctx, final DiskFailoverManager dfoman,
-      RollTrigger t, long checkmillis) {
-    super((S) new LazyOpenDecorator(s));
+  public DiskFailoverDeco(EventSink s, Context ctx,
+      final DiskFailoverManager dfoman, RollTrigger t, long checkmillis) {
+    super(new LazyOpenDecorator<EventSink>(s));
     this.ctx = ctx;
     this.dfoMan = dfoman;
     this.trigger = t;
     this.checkmillis = checkmillis;
   }
 
-  public void setSink(S sink) {
-    this.sink = (S) new LazyOpenDecorator(sink);
+  public void setSink(EventSink sink) {
+    this.sink = new LazyOpenDecorator<EventSink>(sink);
   }
 
   /**
@@ -256,8 +255,8 @@ public class DiskFailoverDeco<S extends EventSink> extends
             "Context does not have a logical node name");
         DiskFailoverManager dfoman = node.getAddDFOManager(dfonode);
 
-        return new DiskFailoverDeco<EventSink>(null, context, dfoman,
-            new TimeTrigger(new ProcessTagger(), delayMillis), checkmillis);
+        return new DiskFailoverDeco(null, context, dfoman, new TimeTrigger(
+            new ProcessTagger(), delayMillis), checkmillis);
       }
     };
   }
diff --git a/src/java/com/cloudera/flume/agent/durability/NaiveFileWALDeco.java b/src/java/com/cloudera/flume/agent/durability/NaiveFileWALDeco.java
index 4cdd1e5..191ebfd 100644
--- a/src/java/com/cloudera/flume/agent/durability/NaiveFileWALDeco.java
+++ b/src/java/com/cloudera/flume/agent/durability/NaiveFileWALDeco.java
@@ -50,14 +50,13 @@ import com.google.common.base.Preconditions;
  * has a subordinate thread that drains the events that have been written to
  * disk. Latches are used to maintain open and close semantics.
  */
-public class NaiveFileWALDeco<S extends EventSink> extends
-    EventSinkDecorator<S> {
+public class NaiveFileWALDeco extends EventSinkDecorator<EventSink> {
   static final Logger LOG = LoggerFactory.getLogger(NaiveFileWALDeco.class);
 
   final WALManager walman;
   final RollTrigger trigger;
   final AckListener queuer;
-  final EventSinkDecorator<S> drainSink;
+  final EventSinkDecorator<EventSink> drainSink;
   final long checkMs;
 
   RollSink input;
@@ -70,7 +69,7 @@ public class NaiveFileWALDeco<S extends EventSink> extends
   CountDownLatch started = null; // blocks open until subthread is started
   volatile IOException lastExn = null;
 
-  public NaiveFileWALDeco(Context ctx, S s, final WALManager walman,
+  public NaiveFileWALDeco(Context ctx, EventSink s, final WALManager walman,
       RollTrigger t, AckListener al, long checkMs) {
     super(s);
     this.ctx = ctx;
@@ -78,9 +77,8 @@ public class NaiveFileWALDeco<S extends EventSink> extends
     this.trigger = t;
     this.queuer = new AckListener.Empty();
     this.al = al;
-    // TODO get rid of this cast.
-    this.drainSink = (EventSinkDecorator<S>) new EventSinkDecorator(
-        new LazyOpenDecorator(new AckChecksumRegisterer<S>(s, al)));
+    this.drainSink = new LazyOpenDecorator<EventSink>(
+        new AckChecksumRegisterer<EventSink>(s, al));
     this.checkMs = checkMs;
   }
 
@@ -276,11 +274,11 @@ public class NaiveFileWALDeco<S extends EventSink> extends
     isOpen.set(true);
   }
 
-  public void setSink(S sink) {
+  public void setSink(EventSink sink) {
     this.sink = sink;
     // TODO get rid of this cast.
-    this.drainSink.setSink((S) new LazyOpenDecorator<EventSink>(
-        new AckChecksumRegisterer<S>(sink, al)));
+    this.drainSink.setSink(new LazyOpenDecorator<EventSink>(
+        new AckChecksumRegisterer<EventSink>(sink, al)));
   }
 
   public synchronized boolean rotate() throws InterruptedException {
@@ -325,9 +323,8 @@ public class NaiveFileWALDeco<S extends EventSink> extends
         // TODO (jon) this is going to be unsafe because it creates before open.
         // This needs to be pushed into the logic of the decorator
         WALManager walman = node.getAddWALManager(walnode);
-        return new NaiveFileWALDeco<EventSink>(context, null, walman,
-            new TimeTrigger(delayMillis), node.getAckChecker()
-                .getAgentAckQueuer(), checkMs);
+        return new NaiveFileWALDeco(context, null, walman, new TimeTrigger(
+            delayMillis), node.getAckChecker().getAgentAckQueuer(), checkMs);
       }
     };
   }
diff --git a/src/javatest/com/cloudera/flume/agent/TestEndToEndAckFailureRecovery.java b/src/javatest/com/cloudera/flume/agent/TestEndToEndAckFailureRecovery.java
index 26287f6..b714b26 100644
--- a/src/javatest/com/cloudera/flume/agent/TestEndToEndAckFailureRecovery.java
+++ b/src/javatest/com/cloudera/flume/agent/TestEndToEndAckFailureRecovery.java
@@ -120,9 +120,9 @@ public class TestEndToEndAckFailureRecovery {
     Context ctx = new ReportTestingContext();
     EventSink es = FlumeBuilder.buildSink(ctx, snk);
     // Excessively large timeout because I am manually triggering it.
-    NaiveFileWALDeco<EventSink> wal = new NaiveFileWALDeco<EventSink>(ctx, es,
-        node.getWalManager(), new TimeTrigger(new ProcessTagger(), 100000),
-        node.getAckChecker().getAgentAckQueuer(), 1000);
+    NaiveFileWALDeco wal = new NaiveFileWALDeco(ctx, es, node.getWalManager(),
+        new TimeTrigger(new ProcessTagger(), 100000), node.getAckChecker()
+            .getAgentAckQueuer(), 1000);
 
     // open and send data.
     wal.open();
@@ -185,9 +185,9 @@ public class TestEndToEndAckFailureRecovery {
     Context ctx = new ReportTestingContext();
     EventSink es = FlumeBuilder.buildSink(ctx, snk);
 
-    NaiveFileWALDeco<EventSink> wal = new NaiveFileWALDeco<EventSink>(ctx, es,
-        node.getWalManager(), new TimeTrigger(new ProcessTagger(), 100000),
-        node.getAckChecker().getAgentAckQueuer(), 1000);
+    NaiveFileWALDeco wal = new NaiveFileWALDeco(ctx, es, node.getWalManager(),
+        new TimeTrigger(new ProcessTagger(), 100000), node.getAckChecker()
+            .getAgentAckQueuer(), 1000);
     wal.open();
 
     // // Phase I - (20 events arrive, 3 groups make it)
@@ -296,9 +296,9 @@ public class TestEndToEndAckFailureRecovery {
     Context ctx = new ReportTestingContext();
     EventSink es = FlumeBuilder.buildSink(ctx, snk);
 
-    NaiveFileWALDeco<EventSink> wal = new NaiveFileWALDeco<EventSink>(ctx, es,
-        node.getWalManager(), new TimeTrigger(new ProcessTagger(), 100000),
-        node.getAckChecker().getAgentAckQueuer(), 1000);
+    NaiveFileWALDeco wal = new NaiveFileWALDeco(ctx, es, node.getWalManager(),
+        new TimeTrigger(new ProcessTagger(), 100000), node.getAckChecker()
+            .getAgentAckQueuer(), 1000);
     wal.open();
 
     // // Phase I - (19 events arrive, 3 groups make it)
@@ -382,9 +382,9 @@ public class TestEndToEndAckFailureRecovery {
     EventSink es = FlumeBuilder.buildSink(ctx, snk);
 
     // Big delay values so that test has to force different actions.
-    NaiveFileWALDeco<EventSink> wal = new NaiveFileWALDeco<EventSink>(ctx, es,
-        node.getWalManager(), new TimeTrigger(new ProcessTagger(), 100000),
-        node.getAckChecker().getAgentAckQueuer(), 1000);
+    NaiveFileWALDeco wal = new NaiveFileWALDeco(ctx, es, node.getWalManager(),
+        new TimeTrigger(new ProcessTagger(), 100000), node.getAckChecker()
+            .getAgentAckQueuer(), 1000);
     wal.open();
 
     // Phase I (20 message sent, 3 groups good, 1 group bad)
@@ -449,9 +449,9 @@ public class TestEndToEndAckFailureRecovery {
     String snk = "{ ackChecker => [ console, counter(\"" + rpt + "\") ] }";
     Context ctx = new ReportTestingContext();
     EventSink es = FlumeBuilder.buildSink(ctx, snk);
-    NaiveFileWALDeco<EventSink> wal = new NaiveFileWALDeco<EventSink>(ctx, es,
-        node.getWalManager(), new TimeTrigger(new ProcessTagger(), 100), node
-            .getAckChecker().getAgentAckQueuer(), 10000);
+    NaiveFileWALDeco wal = new NaiveFileWALDeco(ctx, es, node.getWalManager(),
+        new TimeTrigger(new ProcessTagger(), 100), node.getAckChecker()
+            .getAgentAckQueuer(), 10000);
 
     // open and send data.
     wal.open();
diff --git a/src/javatest/com/cloudera/flume/agent/diskfailover/TestConcurrentDFOMan.java b/src/javatest/com/cloudera/flume/agent/diskfailover/TestConcurrentDFOMan.java
index 613759c..4afeb18 100644
--- a/src/javatest/com/cloudera/flume/agent/diskfailover/TestConcurrentDFOMan.java
+++ b/src/javatest/com/cloudera/flume/agent/diskfailover/TestConcurrentDFOMan.java
@@ -155,9 +155,8 @@ public class TestConcurrentDFOMan {
             dfos[idx] = dfoMan; // save for checking.
 
             // short trigger causes lots of rolls
-            EventSink snk = new DiskFailoverDeco<EventSink>(cnt1,
-                LogicalNodeContext.testingContext(), dfoMan, new TimeTrigger(
-                    100), 50);
+            EventSink snk = new DiskFailoverDeco(cnt1, LogicalNodeContext
+                .testingContext(), dfoMan, new TimeTrigger(100), 50);
 
             ReportManager.get().add(cnt1);
             // make each parallel instance send a slightly different number of
@@ -193,7 +192,9 @@ public class TestConcurrentDFOMan {
           .getReportable("count." + i);
       // check for the slightly different counts based on thread.
       int exp = events + i;
-      LOG.info("count." + i + " expected " + exp + " and got " + cnt.getCount());
+      LOG
+          .info("count." + i + " expected " + exp + " and got "
+              + cnt.getCount());
       assertEquals(exp, (int) cnt.getCount());
 
       // check dfo reports to see if they are sane.
diff --git a/src/javatest/com/cloudera/flume/agent/diskfailover/TestDiskFailoverDeco.java b/src/javatest/com/cloudera/flume/agent/diskfailover/TestDiskFailoverDeco.java
index 0837a91..c892770 100644
--- a/src/javatest/com/cloudera/flume/agent/diskfailover/TestDiskFailoverDeco.java
+++ b/src/javatest/com/cloudera/flume/agent/diskfailover/TestDiskFailoverDeco.java
@@ -78,13 +78,12 @@ public class TestDiskFailoverDeco {
     doNothing().when(msrc).open();
     doNothing().when(msrc).close();
 
-    when(msrc.next()).thenReturn(e1).thenReturn(e2).thenReturn(e3)
-        .thenReturn(null);
+    when(msrc.next()).thenReturn(e1).thenReturn(e2).thenReturn(e3).thenReturn(
+        null);
 
-    DiskFailoverDeco<EventSink> snk = new DiskFailoverDeco<EventSink>(msnk,
-        LogicalNodeContext.testingContext(), FlumeNode.getInstance()
-            .getDFOManager(), new TimeTrigger(new ProcessTagger(), 60000),
-        300000);
+    DiskFailoverDeco snk = new DiskFailoverDeco(msnk, LogicalNodeContext
+        .testingContext(), FlumeNode.getInstance().getDFOManager(),
+        new TimeTrigger(new ProcessTagger(), 60000), 300000);
     snk.open();
     try {
       EventUtil.dumpAll(msrc, snk);
diff --git a/src/javatest/com/cloudera/flume/agent/durability/TestConcurrentWALMan.java b/src/javatest/com/cloudera/flume/agent/durability/TestConcurrentWALMan.java
index 3303e1f..27db9eb 100644
--- a/src/javatest/com/cloudera/flume/agent/durability/TestConcurrentWALMan.java
+++ b/src/javatest/com/cloudera/flume/agent/durability/TestConcurrentWALMan.java
@@ -135,8 +135,8 @@ public class TestConcurrentWALMan {
             AckChecksumChecker<EventSink> chk = new AckChecksumChecker<EventSink>(
                 cnt1);
             NaiveFileWALManager wman1 = new NaiveFileWALManager(f1);
-            EventSink snk = new NaiveFileWALDeco<EventSink>(new Context(), chk,
-                wman1, new TimeTrigger(1000000), new AckListener.Empty(), 100);
+            EventSink snk = new NaiveFileWALDeco(new Context(), chk, wman1,
+                new TimeTrigger(1000000), new AckListener.Empty(), 100);
 
             ReportManager.get().add(cnt1);
             // make each parallel instance send a slightly different number of
diff --git a/src/javatest/com/cloudera/flume/agent/durability/TestNaiveFileWALDeco.java b/src/javatest/com/cloudera/flume/agent/durability/TestNaiveFileWALDeco.java
index a0e4214..6ff1fe4 100644
--- a/src/javatest/com/cloudera/flume/agent/durability/TestNaiveFileWALDeco.java
+++ b/src/javatest/com/cloudera/flume/agent/durability/TestNaiveFileWALDeco.java
@@ -32,7 +32,6 @@ import org.junit.Before;
 import org.junit.Test;
 
 import com.cloudera.flume.agent.FlumeNode;
-import com.cloudera.flume.conf.Context;
 import com.cloudera.flume.conf.FlumeBuilder;
 import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.conf.LogicalNodeContext;
@@ -81,8 +80,8 @@ public class TestNaiveFileWALDeco {
     // file with ack begin, data, and end messages
     File acked = new File("src/data/acked.00000000.20100204-015814430-0800.seq");
     // Assumes the NaiveFileWALManager!
-    File writing = new File(new File(tmp,
-        BenchmarkHarness.node.getPhysicalNodeName()), "writing");
+    File writing = new File(new File(tmp, BenchmarkHarness.node
+        .getPhysicalNodeName()), "writing");
     writing.mkdirs();
 
     // Must rename file because that name is in the meta data of the event
@@ -138,8 +137,8 @@ public class TestNaiveFileWALDeco {
     // file with ack begin, data, and end messages
     File acked = new File("src/data/acked.00000000.20100204-015814430-0800.seq");
     // Assumes the NaiveFileWALManager!
-    File writing = new File(new File(tmp,
-        BenchmarkHarness.node.getPhysicalNodeName()), "writing");
+    File writing = new File(new File(tmp, BenchmarkHarness.node
+        .getPhysicalNodeName()), "writing");
     writing.mkdirs();
 
     // /////////////////////
@@ -175,9 +174,8 @@ public class TestNaiveFileWALDeco {
     // TODO (jon) is this the right behavior? I think assuming no name changes
     // locally is reasonable for now.
 
-    assertTrue(new File(new File(new File(tmp,
-        BenchmarkHarness.node.getPhysicalNodeName()), "sent"), acked.getName())
-        .exists());
+    assertTrue(new File(new File(new File(tmp, BenchmarkHarness.node
+        .getPhysicalNodeName()), "sent"), acked.getName()).exists());
 
     BenchmarkHarness.cleanupLocalWriteDir();
   }
@@ -201,8 +199,8 @@ public class TestNaiveFileWALDeco {
     // file with ack begin, data and then truncated
     File truncated = new File(
         "src/data/truncated.00000000.20100204-015814430-0800.seq");
-    File writing = new File(new File(tmp,
-        BenchmarkHarness.node.getPhysicalNodeName()), "writing");
+    File writing = new File(new File(tmp, BenchmarkHarness.node
+        .getPhysicalNodeName()), "writing");
 
     writing.mkdirs();
     FileUtil.dumbfilecopy(truncated, new File(writing, truncated.getName()));
@@ -253,10 +251,9 @@ public class TestNaiveFileWALDeco {
    */
   @Test
   public void testAppendBeforeOpen() throws InterruptedException {
-    final NaiveFileWALDeco<EventSink> d = new NaiveFileWALDeco<EventSink>(
-        LogicalNodeContext.testingContext(), new NullSink(),
-        new NaiveFileWALManager(new File("/tmp")), new SizeTrigger(0, null),
-        new AckListener.Empty(), 1000000);
+    final NaiveFileWALDeco d = new NaiveFileWALDeco(LogicalNodeContext
+        .testingContext(), new NullSink(), new NaiveFileWALManager(new File(
+        "/tmp")), new SizeTrigger(0, null), new AckListener.Empty(), 1000000);
     final CountDownLatch cdl1 = new CountDownLatch(1);
     new Thread() {
       public void run() {
@@ -284,10 +281,9 @@ public class TestNaiveFileWALDeco {
   @Test
   public void testBadRegistererAppend() throws InterruptedException {
 
-    final NaiveFileWALDeco<EventSink> d = new NaiveFileWALDeco<EventSink>(
-        LogicalNodeContext.testingContext(), new NullSink(),
-        new NaiveFileWALManager(new File("/tmp")), new SizeTrigger(0, null),
-        new AckListener.Empty(), 1000000);
+    final NaiveFileWALDeco d = new NaiveFileWALDeco(LogicalNodeContext
+        .testingContext(), new NullSink(), new NaiveFileWALManager(new File(
+        "/tmp")), new SizeTrigger(0, null), new AckListener.Empty(), 1000000);
 
     final CountDownLatch cdl1 = new CountDownLatch(1);
     new Thread() {
@@ -324,7 +320,7 @@ public class TestNaiveFileWALDeco {
       };
 
       FlumeNode node = FlumeNode.getInstance();
-      EventSinkDecorator<EventSink> deco = new NaiveFileWALDeco<EventSink>(
+      EventSinkDecorator<EventSink> deco = new NaiveFileWALDeco(
           LogicalNodeContext.testingContext(), snk, node.getWalManager(),
           new TimeTrigger(new ProcessTagger(), 1000), node.getAckChecker()
               .getAgentAckQueuer(), 1000);
diff --git a/src/javatest/com/cloudera/flume/agent/durability/TestNaiveFileWALManagerConcurrently.java b/src/javatest/com/cloudera/flume/agent/durability/TestNaiveFileWALManagerConcurrently.java
index 1cf5ec7..bdaae9b 100644
--- a/src/javatest/com/cloudera/flume/agent/durability/TestNaiveFileWALManagerConcurrently.java
+++ b/src/javatest/com/cloudera/flume/agent/durability/TestNaiveFileWALManagerConcurrently.java
@@ -74,7 +74,7 @@ public class TestNaiveFileWALManagerConcurrently {
     Context ctx = new ReportTestingContext();
     EventSink cntsnk = new CompositeSink(ctx, "counter(\"total\")");
     // use the same wal, but different counter.
-    final EventSink snk = new NaiveFileWALDeco<EventSink>(ctx, cntsnk, wal,
+    final EventSink snk = new NaiveFileWALDeco(ctx, cntsnk, wal,
         new TimeTrigger(new ProcessTagger(), 1000000), new AckListener.Empty(),
         1000000);
     snk.open();
@@ -138,9 +138,8 @@ public class TestNaiveFileWALManagerConcurrently {
             EventSink snk = new CompositeSink(ctx, "counter(\"total." + idx
                 + "\")");
             // use the same wal, but different counter.
-            snk = new NaiveFileWALDeco<EventSink>(ctx, snk, wal,
-                new TimeTrigger(new ProcessTagger(), 1000000),
-                new AckListener.Empty(), 1000000);
+            snk = new NaiveFileWALDeco(ctx, snk, wal, new TimeTrigger(
+                new ProcessTagger(), 1000000), new AckListener.Empty(), 1000000);
             src.open();
             snk.open();
 
@@ -207,9 +206,8 @@ public class TestNaiveFileWALManagerConcurrently {
                 + "\")");
             // use the same wal, but different counter.
 
-            snk = new NaiveFileWALDeco<EventSink>(ctx, snk, wal,
-                new TimeTrigger(new ProcessTagger(), 1000000),
-                new AckListener.Empty(), 1000000);
+            snk = new NaiveFileWALDeco(ctx, snk, wal, new TimeTrigger(
+                new ProcessTagger(), 1000000), new AckListener.Empty(), 1000000);
 
             start.await();
 
-- 
1.7.0.4

