@Test
public void testBatchSize() throws IOException
{

  final int batchSize = 3;
  operator = new NiFiSinglePortOutputOperator(stsBuilder, dpBuilder, windowDataManager, batchSize);
  operator.setup(context);
  // get the mock client which will capture each transactions
  final MockSiteToSiteClient mockClient = (MockSiteToSiteClient)operator.client;
  final String tuple1 = "tuple1";
  final String tuple2 = "tuple2";
  final String tuple3 = "tuple3";
  final String tuple4 = "tuple4";
  final String tuple5 = "tuple5";
  operator.beginWindow(1);
  operator.inputPort.process(tuple1);
  Assert.assertEquals(0, mockClient.getMockTransactions().size());
  operator.inputPort.process(tuple2);
  Assert.assertEquals(0, mockClient.getMockTransactions().size());
  // should cause the port to flush and create a transaction
  operator.inputPort.process(tuple3);
  Assert.assertEquals(1, mockClient.getMockTransactions().size());
  operator.inputPort.process(tuple4);
  Assert.assertEquals(1, mockClient.getMockTransactions().size());
  operator.inputPort.process(tuple5);
  Assert.assertEquals(1, mockClient.getMockTransactions().size());
  // should flush tuples 4 and 5 and cause a new transaction
  operator.endNewWindow();
  Assert.assertEquals(2, mockClient.getMockTransactions().size());
  // verify we sent the correct content
  List<String> expectedContents = Arrays.asList(tuple1, tuple2, tuple3, tuple4, tuple5);
  List<MockTransaction> transactions = mockClient.getMockTransactions();
  verifyTransactions(expectedContents, transactions);
}