DataflowSideInputReadCounter implements SideInputReadCounter { @Override public void addBytesRead(long n) { if (currentCounter != null) { currentCounter.addValue(n); } } DataflowSideInputReadCounter(
      DataflowExecutionContext executionContext,
      DataflowOperationContext operationContext,
      int sideInputIndex); @Override void addBytesRead(long n); @Override Closeable enter(); @Override String toString();  }
@Test public void testAddBytesReadUpdatesCounter() { DataflowExecutionContext mockedExecutionContext = mock(DataflowExecutionContext.class); DataflowOperationContext mockedOperationContext = mock(DataflowOperationContext.class); final int siIndexId = 3; ExecutionStateTracker mockedExecutionStateTracker = mock(ExecutionStateTracker.class); when(mockedExecutionContext.getExecutionStateTracker()).thenReturn(mockedExecutionStateTracker); Thread mockedThreadObject = mock(Thread.class); when(mockedExecutionStateTracker.getTrackedThread()).thenReturn(mockedThreadObject); DataflowExecutionState mockedExecutionState = mock(DataflowExecutionState.class); when(mockedExecutionStateTracker.getCurrentState()).thenReturn(mockedExecutionState); NameContext mockedNameContext = mock(NameContext.class); when(mockedExecutionState.getStepName()).thenReturn(mockedNameContext); when(mockedNameContext.originalName()).thenReturn("DummyName"); NameContext mockedDeclaringNameContext = mock(NameContext.class); when(mockedOperationContext.nameContext()).thenReturn(mockedDeclaringNameContext); when(mockedDeclaringNameContext.originalName()).thenReturn("DummyDeclaringName"); CounterFactory mockedCounterFactory = mock(CounterFactory.class); when(mockedExecutionContext.getCounterFactory()).thenReturn(mockedCounterFactory); Counter<Long, Long> mockedCounter = mock(Counter.class); when(mockedCounterFactory.longSum(any())).thenReturn(mockedCounter); when(mockedExecutionContext.getExecutionStateRegistry()) .thenReturn(mock(DataflowExecutionStateRegistry.class)); DataflowSideInputReadCounter testObject = new DataflowSideInputReadCounter(mockedExecutionContext, mockedOperationContext, siIndexId); testObject.addBytesRead(10l); verify(mockedCounter).addValue(10l); }