@Override
  public void addBytesRead(long n) {
    if (currentCounter != null) {
      currentCounter.addValue(n);
    }
  }
@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);
  }