Add 0 value for LLC_PARTITION_CONSUMING gauge whenever we stop consuming
@@ -441,6 +441,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
if (_state.shouldConsume()) {
consumeLoop(); // Consume until we reached the end criteria, or we are stopped.
}
+ _serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.LLC_PARTITION_CONSUMING, 0);
if (_shouldStop) {
break;
}
@@ -750,13 +751,14 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
}
public void goOnlineFromConsuming(RealtimeSegmentZKMetadata metadata) throws InterruptedException {
+ _serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.LLC_PARTITION_CONSUMING, 0);
+ try {
LLCRealtimeSegmentZKMetadata llcMetadata = (LLCRealtimeSegmentZKMetadata)metadata;
// Remove the segment file before we do anything else.
removeSegmentFile();
_leaseExtender.removeSegment(_segmentNameStr);
final long endOffset = llcMetadata.getEndOffset();
- segmentLogger.info("State: {}, transitioning from CONSUMING to ONLINE (startOffset: {}, endOffset: {})",
- _state.toString(), _startOffset, endOffset);
+ segmentLogger.info("State: {}, transitioning from CONSUMING to ONLINE (startOffset: {}, endOffset: {})", _state.toString(), _startOffset, endOffset);
stop();
segmentLogger.info("Consumer thread stopped in state {}", _state.toString());
@@ -801,6 +803,11 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
downloadSegmentAndReplace(llcMetadata);
break;
}
+ } catch (Exception e) {
+ throw(e);
+ } finally {
+ _serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.LLC_PARTITION_CONSUMING, 0);
+ }
}
protected void downloadSegmentAndReplace(LLCRealtimeSegmentZKMetadata metadata) {
@@ -822,6 +829,8 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
// We will end up downloading the segment, so this is not a serious problem
segmentLogger.warn("Exception when catching up to final offset", e);
return false;
+ } finally {
+ _serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.LLC_PARTITION_CONSUMING, 0);
}
if (_currentOffset != endOffset) {
// Timeout?
