Jackie-Jiang commented on code in PR #8663:
URL: https://github.com/apache/pinot/pull/8663#discussion_r877342015
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1121,39 +1132,28 @@ IdealState ensureAllPartitionsConsuming(TableConfig
tableConfig, PartitionLevelS
// 3. we should never end up with some replicas ONLINE and some
OFFLINE.
if (isAllInstancesInState(instanceStateMap,
SegmentStateModel.OFFLINE)) {
LOGGER.info("Repairing segment: {} which is OFFLINE for all
instances in IdealState", latestSegmentName);
-
- // Create a new segment to re-consume from the previous start
offset
- LLCSegmentName newLLCSegmentName =
getNextLLCSegmentName(latestLLCSegmentName, currentTimeMs);
StreamPartitionMsgOffset startOffset =
offsetFactory.create(latestSegmentZKMetadata.getStartOffset());
- StreamPartitionMsgOffset partitionGroupSmallestOffset =
- getPartitionGroupSmallestOffset(streamConfig,
partitionGroupId);
-
- // Start offset must be higher than the start offset of the stream
- if (partitionGroupSmallestOffset.compareTo(startOffset) > 0) {
- LOGGER.error("Data lost from offset: {} to: {} for partition: {}
of table: {}", startOffset,
- partitionGroupSmallestOffset, partitionGroupId,
realtimeTableName);
- _controllerMetrics.addMeteredTableValue(realtimeTableName,
ControllerMeter.LLC_STREAM_DATA_LOSS, 1L);
- startOffset = partitionGroupSmallestOffset;
- }
-
- CommittingSegmentDescriptor committingSegmentDescriptor =
- new CommittingSegmentDescriptor(latestSegmentName,
startOffset.toString(), 0);
- createNewSegmentZKMetadata(tableConfig, streamConfig,
newLLCSegmentName, currentTimeMs,
- committingSegmentDescriptor, latestSegmentZKMetadata,
instancePartitions, numPartitions, numReplicas,
- newPartitionGroupMetadataList);
- String newSegmentName = newLLCSegmentName.getSegmentName();
- updateInstanceStatesForNewConsumingSegment(instanceStatesMap,
null, newSegmentName, segmentAssignment,
- instancePartitionsMap);
+ createNewConsumingSegment(tableConfig, streamConfig,
latestSegmentZKMetadata, currentTimeMs,
+ partitionGroupId, newPartitionGroupMetadataList,
instancePartitions, instanceStatesMap,
+ segmentAssignment, instancePartitionsMap, startOffset);
} else {
if (newPartitionGroupSet.contains(partitionGroupId)) {
- // If we get here, that means in IdealState, the latest segment
has no CONSUMING replicas, but has
- // replicas
- // not OFFLINE. That is an unexpected state which cannot be
fixed by the validation manager currently. In
- // that case, we need to either extend this part to handle the
state, or prevent segments from getting
- // into
- // such state.
- LOGGER
- .error("Got unexpected instance state map: {} for segment:
{}", instanceStateMap, latestSegmentName);
+ // If we get here, that means in IdealState, the latest segment
has all replicas ONLINE, but no
+ // CONSUMING segments.
+ // Create a new IN_PROGRESS segment in PROPERTYSTORE,
+ // add it as CONSUMING segment to IDEALSTATE.
+
+ if (recreateDeletedConsumingSegment &&
Status.DONE.equals(latestSegmentZKMetadata.getStatus())
+ && isAllInstancesInState(instanceStateMap,
SegmentStateModel.ONLINE)
+ ) {
+ StreamPartitionMsgOffset startOffset =
offsetFactory.create(latestSegmentZKMetadata.getEndOffset());
+ createNewConsumingSegment(tableConfig, streamConfig,
latestSegmentZKMetadata, currentTimeMs,
+ partitionGroupId, newPartitionGroupMetadataList,
instancePartitions,
+ instanceStatesMap, segmentAssignment,
instancePartitionsMap, startOffset);
+ } else {
+ LOGGER.error("Got unexpected instance state map: {} for
segment: {}",
+ instanceStateMap, latestSegmentName);
+ }
Review Comment:
(code style) Let's reformat this part
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1121,39 +1132,28 @@ IdealState ensureAllPartitionsConsuming(TableConfig
tableConfig, PartitionLevelS
// 3. we should never end up with some replicas ONLINE and some
OFFLINE.
if (isAllInstancesInState(instanceStateMap,
SegmentStateModel.OFFLINE)) {
LOGGER.info("Repairing segment: {} which is OFFLINE for all
instances in IdealState", latestSegmentName);
-
- // Create a new segment to re-consume from the previous start
offset
- LLCSegmentName newLLCSegmentName =
getNextLLCSegmentName(latestLLCSegmentName, currentTimeMs);
StreamPartitionMsgOffset startOffset =
offsetFactory.create(latestSegmentZKMetadata.getStartOffset());
- StreamPartitionMsgOffset partitionGroupSmallestOffset =
- getPartitionGroupSmallestOffset(streamConfig,
partitionGroupId);
-
- // Start offset must be higher than the start offset of the stream
- if (partitionGroupSmallestOffset.compareTo(startOffset) > 0) {
- LOGGER.error("Data lost from offset: {} to: {} for partition: {}
of table: {}", startOffset,
- partitionGroupSmallestOffset, partitionGroupId,
realtimeTableName);
- _controllerMetrics.addMeteredTableValue(realtimeTableName,
ControllerMeter.LLC_STREAM_DATA_LOSS, 1L);
- startOffset = partitionGroupSmallestOffset;
- }
-
- CommittingSegmentDescriptor committingSegmentDescriptor =
- new CommittingSegmentDescriptor(latestSegmentName,
startOffset.toString(), 0);
- createNewSegmentZKMetadata(tableConfig, streamConfig,
newLLCSegmentName, currentTimeMs,
- committingSegmentDescriptor, latestSegmentZKMetadata,
instancePartitions, numPartitions, numReplicas,
- newPartitionGroupMetadataList);
- String newSegmentName = newLLCSegmentName.getSegmentName();
- updateInstanceStatesForNewConsumingSegment(instanceStatesMap,
null, newSegmentName, segmentAssignment,
- instancePartitionsMap);
+ createNewConsumingSegment(tableConfig, streamConfig,
latestSegmentZKMetadata, currentTimeMs,
+ partitionGroupId, newPartitionGroupMetadataList,
instancePartitions, instanceStatesMap,
+ segmentAssignment, instancePartitionsMap, startOffset);
} else {
if (newPartitionGroupSet.contains(partitionGroupId)) {
- // If we get here, that means in IdealState, the latest segment
has no CONSUMING replicas, but has
- // replicas
- // not OFFLINE. That is an unexpected state which cannot be
fixed by the validation manager currently. In
- // that case, we need to either extend this part to handle the
state, or prevent segments from getting
- // into
- // such state.
- LOGGER
- .error("Got unexpected instance state map: {} for segment:
{}", instanceStateMap, latestSegmentName);
+ // If we get here, that means in IdealState, the latest segment
has all replicas ONLINE, but no
+ // CONSUMING segments.
Review Comment:
(nit)
```suggestion
// If we get here, that means in IdealState, the latest
segment has all replicas ONLINE.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
