@Test
public void testCustomMetricsTransport() throws Exception
{

  TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
  GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);
  dag.addStream("o1.outport", o1.outport, o2.inport1);
  dag.setAttribute(LogicalPlan.METRICS_TRANSPORT, new TestMetricTransport("xyz"));
  StramLocalCluster lc = new StramLocalCluster(dag);
  StreamingContainerManager dnmgr = lc.dnmgr;
  StramAppContext appContext = new StramTestSupport.TestAppContext(dag.getAttributes());
  AppDataPushAgent pushAgent = new AppDataPushAgent(dnmgr, appContext);
  pushAgent.init();
  pushAgent.pushData();
  Assert.assertTrue(TestMetricTransport.messages.size() > 0);
  pushAgent.close();
  String msg = TestMetricTransport.messages.get(0);
  Assert.assertTrue(msg.startsWith("xyz:"));
}