Jackie-Jiang commented on code in PR #8663:
URL: https://github.com/apache/pinot/pull/8663#discussion_r876352201
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1061,6 +1068,9 @@ IdealState ensureAllPartitionsConsuming(TableConfig
tableConfig, PartitionLevelS
// b. update current segment in idealstate to ONLINE (only if partition
is present in newPartitionGroupMetadata)
// c. add new segment in idealstate to CONSUMING on the hosts (only if
partition is present in
// newPartitionGroupMetadata)
+ // 2. The latest metadata is in DONE state, but the idealstate has no
segment in CONSUMING state.
+ // a. Create metadata for new IN_PROGRESS segment with startOffset set
to latest segments' end offset.
+ // b. Add the newly created segment to idealstate with segment state
set to CONSUMING.
// 2. The latest metadata is IN_PROGRESS, but segment is not there in
idealstate.
Review Comment:
(minor) Change the index (currently there are 2 index `2`)
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1121,39 +1131,23 @@ 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, offsetFactory, instanceStatesMap,
+ segmentAssignment, instancePartitionsMap, true);
} else {
if (newPartitionGroupSet.contains(partitionGroupId)) {
// If we get here, that means in IdealState, the latest segment
has no CONSUMING replicas, but has
Review Comment:
```suggestion
// If we get here, that means in IdealState, the latest
segment has all replicas ONLINE
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1211,6 +1205,43 @@ IdealState ensureAllPartitionsConsuming(TableConfig
tableConfig, PartitionLevelS
return idealState;
}
+ private void createNewConsumingSegment(TableConfig tableConfig,
PartitionLevelStreamConfig streamConfig,
Review Comment:
Let's pass in the `startOffset` of the new segment, instead of
`offsetFactory` and `isLatestSegmentOffline`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1121,39 +1131,23 @@ 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, offsetFactory, instanceStatesMap,
+ segmentAssignment, instancePartitionsMap, true);
} else {
if (newPartitionGroupSet.contains(partitionGroupId)) {
// If we get here, that means in IdealState, the latest segment
has no CONSUMING replicas, but has
Review Comment:
Let's add some checks before fixing the segment:
- latestSegmentZKMetadata.getStatus() == Status.DONE
- isAllInstancesInState(instanceStateMap, SegmentStateModel.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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
