public void goOnlineFromConsuming(SegmentZKMetadata segmentZKMetadata)
      throws InterruptedException {
    _serverMetrics.setValueOfTableGauge(_clientId, ServerGauge.LLC_PARTITION_CONSUMING, 0);
    try {
      // Remove the segment file before we do anything else.
      removeSegmentFile();
      _leaseExtender.removeSegment(_segmentNameStr);
      StreamPartitionMsgOffset endOffset = _streamPartitionMsgOffsetFactory.create(segmentZKMetadata.getEndOffset());
      _segmentLogger.info("State: {}, transitioning from CONSUMING to ONLINE (startOffset: {}, endOffset: {})", _state,
          _startOffset, endOffset);
      stop();
      _segmentLogger.info("Consumer thread stopped in state {}", _state);

      switch (_state) {
        case COMMITTED:
        case RETAINED:
          // Nothing to do. we already built local segment and swapped it with in-memory data.
          _segmentLogger.info("State {}. Nothing to do", _state.toString());
          break;
        case DISCARDED:
        case ERROR:
          _segmentLogger.info("State {}. Downloading to replace", _state.toString());
          downloadSegmentAndReplace(segmentZKMetadata);
          break;
        case CATCHING_UP:
        case HOLDING:
        case INITIAL_CONSUMING:
          switch (_segmentCompletionMode) {
            case DOWNLOAD:
              _segmentLogger.info("State {}. CompletionMode {}. Downloading to replace", _state.toString(),
                  _segmentCompletionMode);
              downloadSegmentAndReplace(segmentZKMetadata);
              break;
            case DEFAULT:
              // Allow to catch up upto final offset, and then replace.
              if (_currentOffset.compareTo(endOffset) > 0) {
                // We moved ahead of the offset that is committed in ZK.
                _segmentLogger
                    .warn("Current offset {} ahead of the offset in zk {}. Downloading to replace", _currentOffset,
                        endOffset);
                downloadSegmentAndReplace(segmentZKMetadata);
              } else if (_currentOffset.compareTo(endOffset) == 0) {
                _segmentLogger
                    .info("Current offset {} matches offset in zk {}. Replacing segment", _currentOffset, endOffset);
                buildSegmentAndReplace();
              } else {
                _segmentLogger.info("Attempting to catch up from offset {} to {} ", _currentOffset, endOffset);
                boolean success = catchupToFinalOffset(endOffset,
                    TimeUnit.MILLISECONDS.convert(MAX_TIME_FOR_CONSUMING_TO_ONLINE_IN_SECONDS, TimeUnit.SECONDS));
                if (success) {
                  _segmentLogger.info("Caught up to offset {}", _currentOffset);
                  buildSegmentAndReplace();
                } else {
                  _segmentLogger
                      .info("Could not catch up to offset (current = {}). Downloading to replace", _currentOffset);
                  downloadSegmentAndReplace(segmentZKMetadata);
                }
              }
              break;
            default:
              break;
          }
          break;
        default:
          _segmentLogger.info("Downloading to replace segment while in state {}", _state.toString());
          downloadSegmentAndReplace(segmentZKMetadata);
          break;
      }
    } catch (Exception e) {
      Utils.rethrowException(e);
    } finally {
      _serverMetrics.setValueOfTableGauge(_clientId, ServerGauge.LLC_PARTITION_CONSUMING, 0);
    }
  }
@Test
  public void testOnlineTransitionAfterStop()
      throws Exception {
    SegmentZKMetadata metadata = new SegmentZKMetadata(SEGMENT_NAME_STR);
    final long finalOffsetValue = START_OFFSET_VALUE + 600;
    final LongMsgOffset finalOffset = new LongMsgOffset(finalOffsetValue);
    metadata.setEndOffset(finalOffset.toString());

    try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
      segmentDataManager._stopWaitTimeMs = 0;
      segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.COMMITTED);
      segmentDataManager.goOnlineFromConsuming(metadata);
      Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
      Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
    }

    try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
      segmentDataManager._stopWaitTimeMs = 0;
      segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.RETAINED);
      segmentDataManager.goOnlineFromConsuming(metadata);
      Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
      Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
    }

    try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
      segmentDataManager._stopWaitTimeMs = 0;
      segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.DISCARDED);
      segmentDataManager.goOnlineFromConsuming(metadata);
      Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
      Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
    }

    try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
      segmentDataManager._stopWaitTimeMs = 0;
      segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.ERROR);
      segmentDataManager.goOnlineFromConsuming(metadata);
      Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
      Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
    }

    // If holding, but we have overshot the expected final offset, the download and replace
    try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
      segmentDataManager._stopWaitTimeMs = 0;
      segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.HOLDING);
      segmentDataManager.setCurrentOffset(finalOffsetValue + 1);
      segmentDataManager.goOnlineFromConsuming(metadata);
      Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
      Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
    }

    // If catching up, but we have overshot the expected final offset, the download and replace
    try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
      segmentDataManager._stopWaitTimeMs = 0;
      segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CATCHING_UP);
      segmentDataManager.setCurrentOffset(finalOffsetValue + 1);
      segmentDataManager.goOnlineFromConsuming(metadata);
      Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
      Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
    }

    // If catching up, but we did not get to the final offset, then download and replace
    try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
      segmentDataManager._stopWaitTimeMs = 0;
      segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CATCHING_UP);
      segmentDataManager._consumeOffsets.add(new LongMsgOffset(finalOffsetValue - 1));
      segmentDataManager.goOnlineFromConsuming(metadata);
      Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
      Assert.assertFalse(segmentDataManager._buildAndReplaceCalled);
    }

    // But then if we get to the exact offset, we get to build and replace, not download
    try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager()) {
      segmentDataManager._stopWaitTimeMs = 0;
      segmentDataManager._state.set(segmentDataManager, RealtimeSegmentDataManager.State.CATCHING_UP);
      segmentDataManager._consumeOffsets.add(finalOffset);
      segmentDataManager.goOnlineFromConsuming(metadata);
      Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
      Assert.assertTrue(segmentDataManager._buildAndReplaceCalled);
    }
  }