@Test public void testEndCriteriaChecking() throws Exception { { FakeLLRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(); segmentDataManager._state.set(segmentDataManager, LLRealtimeSegmentDataManager.State.INITIAL_CONSUMING); Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached()); segmentDataManager.setNumRowsIndexed(maxRowsInSegment - 1); Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached()); segmentDataManager.setNumRowsIndexed(maxRowsInSegment); Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached()); Assert.assertEquals(segmentDataManager.getStopReason(), SegmentCompletionProtocol.REASON_ROW_LIMIT); segmentDataManager.destroy(); } { FakeLLRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(); segmentDataManager._state.set(segmentDataManager, LLRealtimeSegmentDataManager.State.INITIAL_CONSUMING); Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached()); _timeNow += maxTimeForSegmentCloseMs + 1; Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached()); replaceRealtimeSegment(segmentDataManager, 10); _timeNow += maxTimeForSegmentCloseMs; Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached()); Assert.assertEquals(segmentDataManager.getStopReason(), SegmentCompletionProtocol.REASON_TIME_LIMIT); segmentDataManager.destroy(); } { FakeLLRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(); segmentDataManager._state.set(segmentDataManager, LLRealtimeSegmentDataManager.State.CATCHING_UP); final long finalOffset = _startOffsetValue + 100; segmentDataManager.setFinalOffset(finalOffset); segmentDataManager.setCurrentOffset(finalOffset - 1); Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached()); segmentDataManager.setCurrentOffset(finalOffset); Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached()); segmentDataManager.destroy(); } { FakeLLRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(); _timeNow += maxTimeForSegmentCloseMs; segmentDataManager._state.set(segmentDataManager, LLRealtimeSegmentDataManager.State.CATCHING_UP); final long finalOffset = _startOffsetValue + 100; segmentDataManager.setFinalOffset(finalOffset); segmentDataManager.setCurrentOffset(finalOffset - 1); Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached()); segmentDataManager.setCurrentOffset(finalOffset); Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached()); segmentDataManager.destroy(); } { FakeLLRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(); _timeNow += 1; segmentDataManager._state.set(segmentDataManager, LLRealtimeSegmentDataManager.State.CONSUMING_TO_ONLINE); segmentDataManager.setConsumeEndTime(_timeNow + 10); final long finalOffset = _startOffsetValue + 100; segmentDataManager.setFinalOffset(finalOffset); segmentDataManager.setCurrentOffset(finalOffset - 1); Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached()); segmentDataManager.setCurrentOffset(finalOffset); Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached()); segmentDataManager.destroy(); } { FakeLLRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(); segmentDataManager._state.set(segmentDataManager, LLRealtimeSegmentDataManager.State.CONSUMING_TO_ONLINE); final long endTime = _timeNow + 10; segmentDataManager.setConsumeEndTime(endTime); final long finalOffset = _startOffsetValue + 100; segmentDataManager.setFinalOffset(finalOffset); segmentDataManager.setCurrentOffset(finalOffset - 1); _timeNow = endTime - 1; Assert.assertFalse(segmentDataManager.invokeEndCriteriaReached()); _timeNow = endTime; Assert.assertTrue(segmentDataManager.invokeEndCriteriaReached()); segmentDataManager.destroy(); } }
public void destroy() { try { stop(); } catch (InterruptedException e) { segmentLogger.error("Could not stop consumer thread"); } _realtimeSegment.destroy(); closeKafkaConsumers(); }
LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager { public void destroy() { try { stop(); } catch (InterruptedException e) { segmentLogger.error("Could not stop consumer thread"); } _realtimeSegment.destroy(); closeKafkaConsumers(); } }
LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager { public void destroy() { try { stop(); } catch (InterruptedException e) { segmentLogger.error("Could not stop consumer thread"); } _realtimeSegment.destroy(); closeKafkaConsumers(); } LLRealtimeSegmentDataManager(RealtimeSegmentZKMetadata segmentZKMetadata, TableConfig tableConfig,
      RealtimeTableDataManager realtimeTableDataManager, String resourceDataDir, IndexLoadingConfig indexLoadingConfig,
      Schema schema, LLCSegmentName llcSegmentName, Semaphore partitionConsumerSemaphore, ServerMetrics serverMetrics); }
LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager { public void destroy() { try { stop(); } catch (InterruptedException e) { segmentLogger.error("Could not stop consumer thread"); } _realtimeSegment.destroy(); closeKafkaConsumers(); } LLRealtimeSegmentDataManager(RealtimeSegmentZKMetadata segmentZKMetadata, TableConfig tableConfig,
      RealtimeTableDataManager realtimeTableDataManager, String resourceDataDir, IndexLoadingConfig indexLoadingConfig,
      Schema schema, LLCSegmentName llcSegmentName, Semaphore partitionConsumerSemaphore, ServerMetrics serverMetrics); void goOnlineFromConsuming(RealtimeSegmentZKMetadata metadata); void destroy(); void stop(); @Override MutableSegment getSegment(); @Override String getSegmentName(); }
LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager { public void destroy() { try { stop(); } catch (InterruptedException e) { segmentLogger.error("Could not stop consumer thread"); } _realtimeSegment.destroy(); closeKafkaConsumers(); } LLRealtimeSegmentDataManager(RealtimeSegmentZKMetadata segmentZKMetadata, TableConfig tableConfig,
      RealtimeTableDataManager realtimeTableDataManager, String resourceDataDir, IndexLoadingConfig indexLoadingConfig,
      Schema schema, LLCSegmentName llcSegmentName, Semaphore partitionConsumerSemaphore, ServerMetrics serverMetrics); void goOnlineFromConsuming(RealtimeSegmentZKMetadata metadata); void destroy(); void stop(); @Override MutableSegment getSegment(); @Override String getSegmentName();  }