@Override
  public Closeable enter() {
    // Only update status from tracked thread to avoid race condition and inconsistent state updates
    if (executionContext.getExecutionStateTracker().getTrackedThread() != Thread.currentThread()) {
      return () -> {};
    }
    updateCurrentStateIfOutdated();
    return executionContext.getExecutionStateTracker().enterState(currentExecutionState);
  }
@Test
  public void testEnterDoesntEnterStateIfCalledFromDifferentThread() {
    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);

    DataflowExecutionStateRegistry mockedExecutionStateRegistry =
        mock(DataflowExecutionStateRegistry.class);
    when(mockedExecutionContext.getExecutionStateRegistry())
        .thenReturn(mockedExecutionStateRegistry);

    DataflowExecutionState mockedCounterExecutionState = mock(DataflowExecutionState.class);
    when(mockedExecutionStateRegistry.getIOState(any(), any(), any(), any(), any(), any()))
        .thenReturn(mockedCounterExecutionState);
    DataflowSideInputReadCounter testObject =
        new DataflowSideInputReadCounter(mockedExecutionContext, mockedOperationContext, siIndexId);

    testObject.enter();

    verify(mockedExecutionStateTracker, never()).enterState(any());
  }