@Test public void testCommitAfterCatchup() throws Exception { FakeLLRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(); LLRealtimeSegmentDataManager.PartitionConsumer consumer = segmentDataManager.createPartitionConsumer(); final LongMsgOffset firstOffset = new LongMsgOffset(_startOffsetValue + 500); final LongMsgOffset catchupOffset = new LongMsgOffset(firstOffset.getOffset() + 10); segmentDataManager._consumeOffsets.add(firstOffset); segmentDataManager._consumeOffsets.add(catchupOffset); final SegmentCompletionProtocol.Response holdResponse1 = new SegmentCompletionProtocol.Response( new SegmentCompletionProtocol.Response.Params() .withStatus(SegmentCompletionProtocol.ControllerResponseStatus.HOLD). withStreamPartitionMsgOffset(firstOffset.toString())); final SegmentCompletionProtocol.Response catchupResponse = new SegmentCompletionProtocol.Response( new SegmentCompletionProtocol.Response.Params() .withStatus(SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP) .withStreamPartitionMsgOffset(catchupOffset.toString())); final SegmentCompletionProtocol.Response holdResponse2 = new SegmentCompletionProtocol.Response( new SegmentCompletionProtocol.Response.Params().withStreamPartitionMsgOffset(catchupOffset.toString()) .withStatus(SegmentCompletionProtocol.ControllerResponseStatus.HOLD)); final SegmentCompletionProtocol.Response commitResponse = new SegmentCompletionProtocol.Response( new SegmentCompletionProtocol.Response.Params().withStreamPartitionMsgOffset(catchupOffset.toString()) .withStatus(SegmentCompletionProtocol.ControllerResponseStatus.COMMIT)); segmentDataManager._responses.add(holdResponse1); segmentDataManager._responses.add(catchupResponse); segmentDataManager._responses.add(holdResponse2); segmentDataManager._responses.add(commitResponse); consumer.run(); Assert.assertTrue(segmentDataManager._responses.isEmpty()); Assert.assertTrue(segmentDataManager._consumeOffsets.isEmpty()); Assert.assertTrue(segmentDataManager._buildSegmentCalled); Assert.assertFalse(segmentDataManager._buildAndReplaceCalled); Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled); Assert.assertTrue(segmentDataManager._commitSegmentCalled); Assert .assertEquals(segmentDataManager._state.get(segmentDataManager), LLRealtimeSegmentDataManager.State.COMMITTED); 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();  }