[partial-upsert] configure early release of _partitionGroupConsumerSemaphore in RealtimeSegmentDataManager #13256

rohityadav1993 · 2024-05-29T08:52:30Z

bugfix upsert

Relates to #13140

This PR configures if we should do an early release of _partitionGroupConsumerSemaphore through closeStreamConsumers() during segment build and replace and segment download and replace.

By skipping to make this call early, we rely on segment.onDestroy() method to release the semaphore which ensures that the new consuming segment does not start consumption until the previous mutable segment is completely replaced during commit.

This helps in avoiding any unnecessary data inconsistencies between replicas for partial upsert tables but also leads to a stopped consumption during segment commit.

New configs:
Use server configuration: "pinot.server.instance.upsert.default.allow.partial.upsert.consumption.during.commit": "true/false" to toggle consumption during commit for partial upsert table. By defualt the consumption during commit will be paused from this commit onwards.

Use table upsert config: allowPartialUpsertConsumptionDuringCommit: true/false toggle consumption during commit for partial upsert table. If the table config is set as true then the server config is ignored.

codecov-commenter · 2024-05-29T09:37:27Z

Codecov Report

Attention: Patch coverage is 33.33333% with 8 lines in your changes missing coverage. Please review.

Project coverage is 61.98%. Comparing base (59551e4) to head (69e28c2).
Report is 624 commits behind head on master.

Files	Patch %	Lines
...a/manager/realtime/RealtimeSegmentDataManager.java	16.66%	4 Missing and 1 partial ⚠️
...ocal/upsert/TableUpsertMetadataManagerFactory.java	0.00%	3 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #13256      +/-   ##
============================================
+ Coverage     61.75%   61.98%   +0.23%     
+ Complexity      207      198       -9     
============================================
  Files          2436     2550     +114     
  Lines        133233   140267    +7034     
  Branches      20636    21803    +1167     
============================================
+ Hits          82274    86945    +4671     
- Misses        44911    46722    +1811     
- Partials       6048     6600     +552

Flag	Coverage Δ
custom-integration1	`<0.01% <0.00%> (-0.01%)`	⬇️
integration	`<0.01% <0.00%> (-0.01%)`	⬇️
integration1	`<0.01% <0.00%> (-0.01%)`	⬇️
integration2	`0.00% <0.00%> (ø)`
java-11	`61.95% <33.33%> (+0.24%)`	⬆️
java-21	`61.87% <33.33%> (+0.24%)`	⬆️
skip-bytebuffers-false	`61.97% <33.33%> (+0.22%)`	⬆️
skip-bytebuffers-true	`61.84% <33.33%> (+34.11%)`	⬆️
temurin	`61.98% <33.33%> (+0.23%)`	⬆️
unittests	`61.98% <33.33%> (+0.23%)`	⬆️
unittests1	`46.55% <33.33%> (-0.35%)`	⬇️
unittests2	`27.66% <0.00%> (-0.07%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

rohityadav1993 · 2024-06-03T04:27:53Z

cc: @Jackie-Jiang , @klsince for review.

tibrewalpratik17 · 2024-06-03T08:16:31Z

pinot-spi/src/main/java/org/apache/pinot/spi/config/table/UpsertConfig.java

+  @JsonPropertyDescription("Whether to pause partial upsert table's partition ingestion during commit")
+  private boolean _pausePartialUpsertPartitionIngestionDuringCommit;
+


@rohityadav1993 can you also add it as a cluster level config where this table-level config will override the cluster-level value? In Uber, we would like to roll it out for almost all tables as data-correctness > data-ingestion-SLAs. For tables where ingestion lag is in order of 10s of minutes there we can make a call to use this table level config.

The suggestion sounds good, do you know how it is being done for any other table config?

you can check BaseTableDataManager#init it takes in InstanceDataManagerConfig var

tibrewalpratik17 · 2024-06-03T08:17:35Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

+    if (!_realtimeTableDataManager.isPartialUpsertEnabled() || !_tableConfig.getUpsertConfig()
+        .isPausePartialUpsertPartitionIngestionDuringCommit()) {


we should check if tableConfig.getUpsertConfig is not null as well

klsince · 2024-06-04T17:07:55Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

@@ -959,7 +959,13 @@ AtomicBoolean getAcquiredConsumerSemaphore() {

  @VisibleForTesting
  SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) {
-    closeStreamConsumers();
+    // for partial upsert tables, do not release _partitionGroupConsumerSemaphore proactively and rely on onDestroy()


nit: looks like offload() closes the consumers

but anyway it looks right to me that the semaphore should be released in the end, when the segment gets replaced

Jackie-Jiang · 2024-06-04T17:36:19Z

pinot-spi/src/main/java/org/apache/pinot/spi/config/table/UpsertConfig.java

@@ -94,6 +94,9 @@ public enum ConsistencyMode {
  @JsonPropertyDescription("Whether to drop out-of-order record")
  private boolean _dropOutOfOrderRecord;

+  @JsonPropertyDescription("Whether to pause partial upsert table's partition ingestion during commit")
+  private boolean _pausePartialUpsertPartitionIngestionDuringCommit;


I think we should pause consumption by default. Consider renaming it to allowPartialUpsertConsumptionDuringCommit and reverse the behavior?

cc @tibrewalpratik17 What's your opinion on enabling the new feature by default?

+1 on making this the default behaviour!
Would also suggest to add a cluster-level and table-level config to enable / disable it.

Making default as enabled sounds good. This is a behaviour change, I also think it makes sense to make this a cluster level property as well so it can be toggled at bulk.

tibrewalpratik17 · 2024-06-04T17:47:21Z

Suggest to add release-notes label to this!

…meSegmentDataManager

klsince · 2024-06-13T17:16:32Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

@@ -1611,6 +1622,10 @@ public RealtimeSegmentDataManager(SegmentZKMetadata segmentZKMetadata, TableConf
      _segmentLogger
          .info("Starting consumption on realtime consuming segment {} maxRowCount {} maxEndTime {}", llcSegmentName,
              _segmentMaxRowCount, new DateTime(_consumeEndTime, DateTimeZone.UTC));
+      _allowPartialUpsertConsumptionDuringCommit =
+          _realtimeTableDataManager.isPartialUpsertEnabled() ? _tableConfig.getUpsertConfig() != null


the config name as allowPartialUpsertConsumptionDuringCommit or default.allow.partial.upsert.consumption.during.commit makes a lot sense and it's false by default.

but this variable name _allowPartialUpsertConsumptionDuringCommit here might be a bit less readable. Here we had to set it true for non-paritial-upsert table. I'd call it _allowConsumptionDuringCommit, then it's more intuitive to leave it as true for non-paritial-upsert table

_allowConsumptionDuringCommit = !_realtimeTableDataManager.isPartialUpsertEnabled()? true : _tableConfig.getUpsertConfig().isAllowPartialUpsertConsumptionDuringCommit() // nit: no need to check getUpsertConfig() is not null as isPartialUpsertEnabled() is true already.

Sounds good. Updated.

tibrewalpratik17 · 2024-06-14T07:17:47Z

...l/src/main/java/org/apache/pinot/segment/local/upsert/TableUpsertMetadataManagerFactory.java

+      // server level config honoured only when table level config is not set to true
+      if (!upsertConfig.isAllowPartialUpsertConsumptionDuringCommit()) {
+        upsertConfig.setAllowPartialUpsertConsumptionDuringCommit(Boolean.parseBoolean(
+            instanceUpsertConfig.getProperty(UPSERT_DEFAULT_ALLOW_PARTIAL_UPSERT_CONSUMPTION_DURING_COMMIT, "false")));


we wanted to keep this true by default right?

Ah my bad! I see we have changed the behaviour from pause to allow. Then this makes sense!

rohityadav1993 marked this pull request as ready for review June 3, 2024 04:27

rohityadav1993 changed the title ~~configure early release of _partitionGroupConsumerSemaphore in RealtimeSegmentDataManager~~ [partial-upsert] configure early release of _partitionGroupConsumerSemaphore in RealtimeSegmentDataManager Jun 3, 2024

tibrewalpratik17 reviewed Jun 3, 2024

View reviewed changes

klsince approved these changes Jun 4, 2024

View reviewed changes

Jackie-Jiang added upsert bugfix labels Jun 4, 2024

Jackie-Jiang reviewed Jun 4, 2024

View reviewed changes

Jackie-Jiang added release-notes Referenced by PRs that need attention when compiling the next release notes Configuration Config changes (addition/deletion/change in behavior) documentation labels Jun 4, 2024

configure early release of _partitionGroupConsumerSemaphore in Realti…

e9002c5

…meSegmentDataManager

rohityadav1993 force-pushed the pu_inconsistency branch from 16eb518 to e9002c5 Compare June 13, 2024 14:37

klsince reviewed Jun 13, 2024

View reviewed changes

rohityadav1993 force-pushed the pu_inconsistency branch from a90c6a1 to 7e87d37 Compare June 13, 2024 18:51

review comments resolution

69e28c2

rohityadav1993 force-pushed the pu_inconsistency branch from 7e87d37 to 69e28c2 Compare June 14, 2024 06:49

tibrewalpratik17 reviewed Jun 14, 2024

View reviewed changes

tibrewalpratik17 approved these changes Jun 14, 2024

View reviewed changes

klsince approved these changes Jun 14, 2024

View reviewed changes

klsince merged commit 6891150 into apache:master Jun 14, 2024
20 checks passed

rohityadav1993 mentioned this pull request Jun 20, 2024

[Bug] Inconsistent data in partial upsert table #13140

Closed

rohityadav1993 mentioned this pull request Jul 24, 2024

Improve InstanceSelector routing logic for new segments #13683

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[partial-upsert] configure early release of _partitionGroupConsumerSemaphore in RealtimeSegmentDataManager #13256

[partial-upsert] configure early release of _partitionGroupConsumerSemaphore in RealtimeSegmentDataManager #13256

rohityadav1993 commented May 29, 2024 •

edited

Loading

codecov-commenter commented May 29, 2024 •

edited

Loading

rohityadav1993 commented Jun 3, 2024

tibrewalpratik17 Jun 3, 2024

rohityadav1993 Jun 3, 2024

tibrewalpratik17 Jun 3, 2024

tibrewalpratik17 Jun 3, 2024

klsince Jun 4, 2024

Jackie-Jiang Jun 4, 2024

Jackie-Jiang Jun 4, 2024

tibrewalpratik17 Jun 4, 2024

rohityadav1993 Jun 5, 2024

tibrewalpratik17 commented Jun 4, 2024

klsince Jun 13, 2024

rohityadav1993 Jun 13, 2024

tibrewalpratik17 Jun 14, 2024

tibrewalpratik17 Jun 14, 2024

		@JsonPropertyDescription("Whether to pause partial upsert table's partition ingestion during commit")
		private boolean _pausePartialUpsertPartitionIngestionDuringCommit;

		if (!_realtimeTableDataManager.isPartialUpsertEnabled() \|\| !_tableConfig.getUpsertConfig()
		.isPausePartialUpsertPartitionIngestionDuringCommit()) {

[partial-upsert] configure early release of _partitionGroupConsumerSemaphore in RealtimeSegmentDataManager #13256

[partial-upsert] configure early release of _partitionGroupConsumerSemaphore in RealtimeSegmentDataManager #13256

Conversation

rohityadav1993 commented May 29, 2024 • edited Loading

codecov-commenter commented May 29, 2024 • edited Loading

Codecov Report

rohityadav1993 commented Jun 3, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tibrewalpratik17 commented Jun 4, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rohityadav1993 commented May 29, 2024 •

edited

Loading

codecov-commenter commented May 29, 2024 •

edited

Loading