@Test public void testNotLeader() throws Exception { FakeLLRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManager(); LLRealtimeSegmentDataManager.PartitionConsumer consumer = segmentDataManager.createPartitionConsumer(); final LongMsgOffset endOffset = new LongMsgOffset(_startOffsetValue + 500); segmentDataManager._consumeOffsets.add(endOffset); final SegmentCompletionProtocol.Response response = new SegmentCompletionProtocol.Response( new SegmentCompletionProtocol.Response.Params().withStreamPartitionMsgOffset(endOffset.toString()) .withStatus(SegmentCompletionProtocol.ControllerResponseStatus.NOT_LEADER)); for (int i = 0; i < 100; i++) { segmentDataManager._responses.add(response); } consumer.run(); Assert.assertTrue(segmentDataManager._responses.isEmpty()); Assert.assertTrue(segmentDataManager._consumeOffsets.isEmpty()); Assert.assertFalse(segmentDataManager._buildAndReplaceCalled); Assert.assertFalse(segmentDataManager._buildSegmentCalled); Assert.assertFalse(segmentDataManager._commitSegmentCalled); Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled); Assert.assertEquals(segmentDataManager._state.get(segmentDataManager), LLRealtimeSegmentDataManager.State.HOLDING); 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();  }