Log segments with no replicas.
This diff enhances the SegmentStatusChecker to log the segments with no replicas. It also makes sure that there is only one thread doing the Segment Status
Check, so that even if there are multiple leadership change notifications, only
one executor is added.
@@ -53,6 +53,7 @@ public class SegmentStatusChecker {
private static final Logger LOGGER = LoggerFactory.getLogger(SegmentStatusChecker.class);
private static final int SegmentCheckerDefaultIntervalSeconds = 120;
private static final int SegmentCheckerDefaultWaitForPushTimeSeconds = 600 * 1000;
+ private static final int MaxOfflineSegmentsToLog = 5;
private static final String CONTROLLER_LEADER_CHANGE = "CONTROLLER LEADER CHANGE";
public static final String ONLINE = "ONLINE";
public static final String ERROR = "ERROR";
@@ -98,7 +99,6 @@ public class SegmentStatusChecker {
private void startThread() {
LOGGER.info("Starting segment status checker");
-
if (_executorService == null) {
_executorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
@@ -108,7 +108,6 @@ public class SegmentStatusChecker {
return thread;
}
});
- }
// Set up an executor that executes segment status tasks periodically
_executorService.scheduleWithFixedDelay(new Runnable() {
@Override
@@ -120,6 +119,9 @@ public class SegmentStatusChecker {
}
}
}, SegmentCheckerDefaultIntervalSeconds, _segmentStatusIntervalSeconds, TimeUnit.SECONDS);
+ } else {
+ LOGGER.warn("SegmentStatusChecker already running. Attempt to start a duplicate thread");
+ }
}
/**
@@ -181,6 +183,7 @@ public class SegmentStatusChecker {
final int nReplicasIdeal = Integer.parseInt(idealState.getReplicas());
int nReplicasExternal = nReplicasIdeal;
int nErrors = 0;
+ int nOffline = 0;
for (String partitionName : idealState.getPartitionSet()) {
int nReplicas = 0;
int nIdeal = 0;
@@ -209,6 +212,10 @@ public class SegmentStatusChecker {
continue;
}
}
+ nOffline++;
+ if (nOffline < MaxOfflineSegmentsToLog) {
+ LOGGER.warn("Segment {} of table {} has no replicas", partitionName, tableName);
+ }
nReplicasExternal = 0;
continue;
}
@@ -221,6 +228,12 @@ public class SegmentStatusChecker {
nErrors++;
}
}
+ if (nReplicas == 0) {
+ if (nOffline < MaxOfflineSegmentsToLog) {
+ LOGGER.warn("Segment {} of table {} has no online replicas", partitionName, tableName);
+ }
+ nOffline++;
+ }
nReplicasExternal = (nReplicasExternal > nReplicas) ? nReplicas : nReplicasExternal;
}
// Synchronization provided by Controller Gauge to make sure that only one thread updates the gauge
@@ -228,6 +241,9 @@ public class SegmentStatusChecker {
nReplicasExternal);
_metricsRegistry.setValueOfTableGauge(tableName, ControllerGauge.SEGMENTS_IN_ERROR_STATE,
nErrors);
+ if (nOffline > 0) {
+ LOGGER.warn("Table {} has {} segments with no online replicas", tableName, nOffline);
+ }
if (nReplicasExternal < nReplicasIdeal) {
LOGGER.warn("Table {} has {} replicas, below replication threshold :{}", tableName,
nReplicasExternal, nReplicasIdeal);
@@ -252,7 +268,7 @@ public class SegmentStatusChecker {
LOGGER.info("Not the leader of this cluster, stopping Status Checker.");
}
} catch (Exception e) {
- LOGGER.error("Caught exception while processing leader change for path {}", path, e);
+ LOGGER.error("Caught exception {} while processing leader change for path {}", e, path);
}
}
