Upgrade pending segments when a concurrent replace happens#15097
Upgrade pending segments when a concurrent replace happens#15097kfaraz wants to merge 8 commits intoapache:masterfrom
Conversation
kfaraz
added
Area - Compaction
Area - Streaming Ingestion
and removed
Area - Ingestion
labels
| * This action differs from {@link SegmentTransactionalInsertAction} as it is used | ||
| * only with APPEND locks and also upgrades segments as needed. | ||
| */ | ||
| public class CommitRealtimeSegmentsAndMetadataAction implements TaskAction<SegmentPublishResult> |
There was a problem hiding this comment.
Could we not re-use SegmentTransactionAppendAction with the metadata being null for batch and the required values for streaming ingestion (similar to the original insert action)?
Is there anything besides the metadata commit that this action does that the transactional append action doesn't?
There was a problem hiding this comment.
Yeah, I was thinking the same. Let me see what we can do.
| "Cannot use action[%s] for task[%s] as it is holding a lock of type[%s] instead of [APPEND].", | ||
| "CommitRealtimeSegmentsAndMetadata", task.getId(), lock.getType() |
There was a problem hiding this comment.
Since this error will find its way to end user, we should avoid using names such as CommitRealtimeSegmentsAndMetadata
| @JsonProperty("checkpointed") boolean checkpointed, | ||
| @JsonProperty("exclusiveStartPartitions") Set<PartitionIdType> exclusiveStartPartitions | ||
| @JsonProperty("exclusiveStartPartitions") Set<PartitionIdType> exclusiveStartPartitions, | ||
| @JsonProperty("taskLockType") TaskLockType taskLockType |
There was a problem hiding this comment.
This should be nullable for backward compatibility reasons. should it not?
| } | ||
|
|
||
| final String dataSource = appendSegments.iterator().next().getDataSource(); | ||
| final Set<DataSegment> upgradedSegments = connector.retryTransaction( |
There was a problem hiding this comment.
| final Set<DataSegment> upgradedSegments = connector.retryTransaction( | |
| final Set<DataSegment> segmentsToUpgrade = connector.retryTransaction( |
|
Closing this PR as these changes have already been merged as a part of #15039 |
3 participants
[WIP]
Description
Changes
TaskLockTypeinSequenceMetadata. The lock type is used to determine the right action type for committing realtime segments.IndexTaskUtils.emitSegmentPublishMetricsIndexTaskUtils.getMessagesFromSavedParseExceptionsPending changes
IndexerMetadataStorageCoordinatorto commit segments, commit metadata and upgrade segments in the same transactioncommitReplaceSegmentsmethod to identify pending segments to upgrade and create upgraded entriesThis PR has: