package com.cloudera.flume.reporter;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import org.junit.Assert;
import org.junit.Test;
import com.cloudera.flume.conf.Context;
import com.cloudera.flume.conf.FlumeBuilder;
import com.cloudera.flume.conf.FlumeSpecException;
import com.cloudera.flume.core.EventSink;
@Test
public void testSimple()
throws FlumeSpecException {
String s = "console";
EventSink sink = FlumeBuilder.buildSink(new Context(), s);
Map<String, ReportEvent> reports = new HashMap<String, ReportEvent>();
sink.getReports("X.", reports);
String r = "";
for (Entry<String, ReportEvent> e : reports.entrySet()) {
r += e.getKey() + " = " + e.getValue().toText();
}
System.out.println(r);
Assert.assertTrue(r.contains("X." + sink.getName()));
}
@Test
String s = "{ stubbornAppend => console}";
EventSink sink = FlumeBuilder.buildSink(new Context(), s);
Map<String, ReportEvent> reports = new HashMap<String, ReportEvent>();
sink.getReports("X.", reports);
String r = "";
for (Entry<String, ReportEvent> e : reports.entrySet()) {
r += e.getKey() + " = " + e.getValue().toText();
}
System.out.println(r);
Assert.assertTrue(r.contains("X." + sink.getName()));
}
@Test
String s =
"{ intervalSampler(5) => { stubbornAppend => { insistentOpen => console } } }";
EventSink sink = FlumeBuilder.buildSink(new Context(), s);
Map<String, ReportEvent> reports = new HashMap<String, ReportEvent>();
sink.getReports("X.", reports);
String r = "";
for (Entry<String, ReportEvent> e : reports.entrySet()) {
r += e.getKey() + " = " + e.getValue().toText();
}
System.out.println(r);
Assert.assertTrue(r.contains("X." + sink.getName()));
Assert.assertTrue(r.contains("X.IntervalSampler.StubbornAppend.InsistentOpen"));
}
@Test
String s =
"{ ackedWriteAhead => { stubbornAppend => { insistentOpen => console } } }";
EventSink sink = FlumeBuilder.buildSink(new Context(), s);
Map<String, ReportEvent> reports = new HashMap<String, ReportEvent>();
sink.getReports("X.", reports);
String r = "";
for (Entry<String, ReportEvent> e : reports.entrySet()) {
r += e.getKey() + " = " + e.getValue().toText();
}
System.out.println(r);
Assert.assertTrue(r.contains("X." + sink.getName()));
Assert.assertTrue(r.contains("X.NaiveFileWAL.StubbornAppend.InsistentOpen"));
}
@Test
String s =
"{ insistentOpen => { ackedWriteAhead => { stubbornAppend => { insistentOpen => console } } } }";
EventSink sink = FlumeBuilder.buildSink(new Context(), s);
Map<String, ReportEvent> reports = new HashMap<String, ReportEvent>();
sink.getReports("X.", reports);
String r = "";
for (Entry<String, ReportEvent> e : reports.entrySet()) {
r += e.getKey() + " = " + e.getValue().toText();
}
System.out.println(r);
Assert.assertTrue(r.contains("X." + sink.getName()));
Assert.assertTrue(r
.contains("X.InsistentOpen.NaiveFileWAL.StubbornAppend.InsistentOpen"));
}
@Test
String s = " { ackedWriteAhead => < thriftSink ? console > } ";
EventSink sink = FlumeBuilder.buildSink(new Context(), s);
Map<String, ReportEvent> reports = new HashMap<String, ReportEvent>();
sink.getReports("X.", reports);
String r = "";
for (Entry<String, ReportEvent> e : reports.entrySet()) {
r += e.getKey() + " = " + e.getValue().toText();
}
System.out.println(r);
Assert.assertTrue(r.contains("X." + sink.getName()));
Assert.assertTrue(r
.contains("X.NaiveFileWAL.BackoffFailover.primary.ThriftEventSink"));
Assert.assertTrue(r
.contains("X.NaiveFileWAL.BackoffFailover.backup.ConsoleEventSink"));
}
@Test
String s = " { ackedWriteAhead => [ thriftSink , console ] } ";
EventSink sink = FlumeBuilder.buildSink(new Context(), s);
Map<String, ReportEvent> reports = new HashMap<String, ReportEvent>();
sink.getReports("X.", reports);
String r = "";
for (Entry<String, ReportEvent> e : reports.entrySet()) {
r += e.getKey() + " = " + e.getValue().toText();
}
System.out.println(r);
Assert.assertTrue(r.contains("X." + sink.getName()));
Assert.assertTrue(r.contains("X.NaiveFileWAL.Fanout.0.ThriftEventSink"));
Assert.assertTrue(r.contains("X.NaiveFileWAL.Fanout.1.ConsoleEventSink"));
}
}