c2Src.open();
// create agentDFOChain sink
File tmpDir = FileUtil.mktempdir();
String spec = "agentDFOChain(\"localhost:1236\", \"localhost:1237\")";
CompositeSink snk = new CompositeSink(new LogicalNodeContext(
tmpDir.getName(), tmpDir.getName()), spec);
snk.open();
Event e1 = new EventImpl("test 1".getBytes());
Event e2 = new EventImpl("test 2".getBytes());
Event e3 = new EventImpl("test 3".getBytes());
Event e4 = new EventImpl("test 4".getBytes());
// Everything is on and we send some messages.
snk.append(e1);
Clock.sleep(100);
LOG.info("DFO failchain all on metrics {}", c1Src.getMetrics().toString());
assertEquals(1,
(long) c1Src.getMetrics().getLongMetric(ThriftEventSource.A_ENQUEUED));
// it got through, yay.
c1Src.next();
c1Src.close();
// Killed the first of the chain, should go to backup
// the number of events lost here is not consistent after close. this
// seems time based, and the first two seem to be lost
snk.append(e1);
Clock.sleep(20);
snk.append(e2);
Clock.sleep(20);
snk.append(e3);
Clock.sleep(20);
snk.append(e4);
Clock.sleep(20);
LOG.info("DFO failchain kill primary, show 2ndary metrics {}", c2Src
.getMetrics().toString());
long enqueued = (long) c2Src.getMetrics().getLongMetric(
ThriftEventSource.A_ENQUEUED);
assertTrue(
"Expected at least 2 events still enqueueed but got " + enqueued,
2 <= enqueued);
// 2 lost in network buffer, but two received in backup. yay.
c2Src.next();
c2Src.next();
c2Src.close();
e1 = new EventImpl("test 1 b".getBytes());
e2 = new EventImpl("test 2 b".getBytes());
e3 = new EventImpl("test 3 b".getBytes());
e4 = new EventImpl("test 4 b".getBytes());
// all thrift sinks are closed now, we should end up in dfo
snk.append(e1); // lost in thrift sink buffer
Clock.sleep(20);
snk.append(e2); // lost in thrift sink buffer
Clock.sleep(20);
snk.append(e3); // written
Clock.sleep(20);
snk.append(e4); // written
Clock.sleep(20);
LOG.info("DFO Primary returns metrics: {}", snk.getMetrics().toText());
long writingEvts = (long) ReportUtil.getFlattenedReport(snk).getLongMetric(
"backup.DiskFailover.NaiveDiskFailover.writingEvts");
assertTrue("Expected at least 2 writing evetns but had " + writingEvts,
2 <= writingEvts);
e1 = new EventImpl("test 1 c".getBytes());
e2 = new EventImpl("test 2 c".getBytes());
e3 = new EventImpl("test 3 c".getBytes());
e4 = new EventImpl("test 4 c".getBytes());
// re-open destination 1
c1Src.open();
snk.append(e1);
Clock.sleep(20);
snk.append(e2);
Clock.sleep(20);
c1Src.next();
// get handle to roller in dfo log roller to provide data
AgentFailChainSink afcs = (AgentFailChainSink) snk.getSink();
BackOffFailOverSink bofos = (BackOffFailOverSink) ((CompositeSink) afcs.snk)
.getSink();
DiskFailoverDeco dfo = (DiskFailoverDeco) bofos.getBackup();
RollSink dfoWriter = dfo.getDFOWriter();
dfoWriter.rotate(); // allow dfo retry thread to go.