This is an automated email from the ASF dual-hosted git repository.
KKcorps pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 85db332cfca Reject consuming during download for upsert tables that
revert metadata in PROTECTED mode (#19710)
85db332cfca is described below
commit 85db332cfca365508212e712f6527d959e2ed172
Author: Kartik Khare <[email protected]>
AuthorDate: Tue Oct 6 17:01:53 2026 +0530
Reject consuming during download for upsert tables that revert metadata in
PROTECTED mode (#19710)
* Reject consuming during download for upsert tables that revert metadata
in PROTECTED mode
In PROTECTED mode, partial upsert and out-of-order tables revert keys the
replica consumed past the
committed end offset when the downloaded segment replaces the consuming
one. If the next consuming
segment starts during that download, its validDocIds snapshot runs before
the revert and misses the
rows the revert restores. Reject such policies in table config validation.
When the combination still
happens at runtime, the server logs an error and bumps a meter instead of
changing the policy.
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01W5J32DeWRrd7a1e3z6Y5bN
* Rename the upsert revert consumption check and name the rejected setting
Address review: rename to validateConsumptionDuringUpsertRevert and start
the error with the policy or
flag that was rejected, so ALLOW_ALWAYS and ALLOW_DURING_DOWNLOAD_ONLY are
visible in the message.
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01W5J32DeWRrd7a1e3z6Y5bN
* Reject ALLOW_DURING_BUILD_ONLY for upsert revert tables and flag the
build-to-download fallback
A local build under ALLOW_DURING_BUILD_ONLY releases the consumer semaphore
before the CRC check. On a failed build
or a CRC mismatch the server then downloads the committed copy and reverts
upsert metadata while the next consuming
segment can already be taking its snapshot. Validation now accepts only
DISALLOW_ALWAYS for these tables in
PROTECTED mode, except pauseless tables, which consume during build by
design. The server meter now also fires on
that fallback download.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Kartik Khare <[email protected]>
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../apache/pinot/common/metrics/ServerMeter.java | 1 +
.../realtime/RealtimeSegmentDataManager.java | 16 +++-
.../realtime/RealtimeSegmentDataManagerTest.java | 103 ++++++++++++++++++++-
.../upsert/BasePartitionUpsertMetadataManager.java | 1 +
.../upsert/PartitionUpsertMetadataManager.java | 5 +
.../segment/local/utils/TableConfigUtils.java | 32 +++++++
...rrentMapPartitionUpsertMetadataManagerTest.java | 22 +++++
.../segment/local/utils/TableConfigUtilsTest.java | 88 ++++++++++++++++++
8 files changed, 264 insertions(+), 4 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
index 91ca1327069..f862ed63aea 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
@@ -74,6 +74,7 @@ public enum ServerMeter implements AbstractMetrics.Meter {
UPSERT_MISSED_VALID_DOC_ID_SNAPSHOT_COUNT("segments", false),
UPSERT_MISSED_QUERYABLE_DOC_ID_SNAPSHOT_COUNT("segments", false),
UPSERT_PRELOAD_FAILURE("count", false),
+ UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD("segments", false),
ROWS_WITH_ERRORS("rows", false),
LLC_CONTROLLER_RESPONSE_NOT_SENT("messages", true),
LLC_CONTROLLER_RESPONSE_COMMIT("messages", true),
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
index f096ab6cc92..b330c781dff 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -1751,7 +1751,21 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata)
throws Exception {
- if (_parallelSegmentConsumptionPolicy.isAllowedDuringDownload()) {
+ boolean allowedDuringDownload =
_parallelSegmentConsumptionPolicy.isAllowedDuringDownload();
+ // A local build that failed or mismatched the committed CRC has already
released the semaphore when the policy
+ // allows consumption during build, so the next consuming segment can be
running during this download as well
+ boolean releasedDuringBuild =
+ _parallelSegmentConsumptionPolicy.isAllowedDuringBuild() &&
!_consumerSemaphoreAcquired.get();
+ if ((allowedDuringDownload || releasedDuringBuild) &&
_partitionUpsertMetadataManager != null
+ &&
_partitionUpsertMetadataManager.shouldRevertMetadataOnInconsistency(_realtimeSegment))
{
+ // Table config validation rejects this in PROTECTED mode. It still
happens when the mode is switched after the
+ // table is created, when the server-level default allows partial upsert
consumption during commit, or when a
+ // pauseless table falls back to a download after its local build fails.
+ _segmentLogger.error("Next consuming segment can run during download
while this replace reverts upsert "
+ + "metadata. Its snapshot can miss the rows the revert restores. Use
DISALLOW_ALWAYS");
+ _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L);
+ }
+ if (allowedDuringDownload) {
closeStreamConsumerAndReleaseSemaphore();
}
_realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
index c778794ffb8..6dd607bb91c 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
@@ -40,6 +40,7 @@ import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
import org.apache.helix.HelixManager;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.common.utils.LLCSegmentName;
@@ -56,6 +57,7 @@ import
org.apache.pinot.segment.local.segment.creator.Fixtures;
import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager;
import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.local.utils.ServerReloadJobStatusCache;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
@@ -63,6 +65,7 @@ import
org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import
org.apache.pinot.spi.config.table.ingestion.ParallelSegmentConsumptionPolicy;
import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -81,6 +84,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.same;
@@ -233,6 +237,14 @@ public class RealtimeSegmentDataManagerTest {
private FakeRealtimeSegmentDataManager createFakeSegmentManager(boolean
noUpsert, TimeSupplier timeSupplier,
@Nullable String maxRows, @Nullable String maxDuration, @Nullable
TableConfig tableConfig)
throws Exception {
+ return createFakeSegmentManager(noUpsert, timeSupplier, maxRows,
maxDuration, tableConfig,
+ new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry()));
+ }
+
+ private FakeRealtimeSegmentDataManager createFakeSegmentManager(boolean
noUpsert, TimeSupplier timeSupplier,
+ @Nullable String maxRows, @Nullable String maxDuration, @Nullable
TableConfig tableConfig,
+ ServerMetrics serverMetrics)
+ throws Exception {
SegmentZKMetadata segmentZKMetadata = createZkMetadata();
if (tableConfig == null) {
tableConfig = createTableConfig();
@@ -257,7 +269,6 @@ public class RealtimeSegmentDataManagerTest {
_partitionGroupIdToConsumerCoordinatorMap.putIfAbsent(PARTITION_GROUP_ID,
new ConsumerCoordinator(false, tableDataManager));
Schema schema = Fixtures.createSchema();
- ServerMetrics serverMetrics = new
ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry());
return new FakeRealtimeSegmentDataManager(segmentZKMetadata, tableConfig,
tableDataManager,
new File(TEMP_DIR, REALTIME_TABLE_NAME).getAbsolutePath(), schema,
llcSegmentName,
_partitionGroupIdToConsumerCoordinatorMap, serverMetrics,
timeSupplier);
@@ -879,12 +890,77 @@ public class RealtimeSegmentDataManagerTest {
return metadata;
}
+ @Test
+ public void
testDownloadAfterLocalBuildFlagsUpsertRevertWhenConsumingDuringBuild()
+ throws Exception {
+ long finalOffsetValue = START_OFFSET_VALUE + 600;
+
+ // ALLOW_DURING_BUILD_ONLY: the local build releases the consumer
semaphore, its CRC mismatches the committed copy,
+ // and the download that follows reverts upsert metadata while the next
consuming segment can already be running.
+ _partitionGroupIdToConsumerCoordinatorMap.remove(PARTITION_GROUP_ID);
+ ServerMetrics serverMetrics = spy(new
ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry()));
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManagerForUpsertRevert(
+ ParallelSegmentConsumptionPolicy.ALLOW_DURING_BUILD_ONLY,
serverMetrics)) {
+ Semaphore semaphore =
_partitionGroupIdToConsumerCoordinatorMap.get(PARTITION_GROUP_ID).getSemaphore();
+ Assert.assertTrue(semaphore.tryAcquire());
+ segmentDataManager._releaseSemaphoreOnBuild = true;
+ runGoOnlineForCrcGuard(segmentDataManager, crcMetadata(finalOffsetValue,
12345L), false, finalOffsetValue);
+ Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
+ Assert.assertEquals(semaphore.availablePermits(), 1);
+ verify(serverMetrics).addMeteredTableValue(anyString(),
+ eq(ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD),
eq(1L));
+ }
+
+ // ALLOW_DURING_BUILD_ONLY without a local build: the download still holds
the semaphore, so nothing is flagged.
+ _partitionGroupIdToConsumerCoordinatorMap.remove(PARTITION_GROUP_ID);
+ serverMetrics = spy(new
ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry()));
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManagerForUpsertRevert(
+ ParallelSegmentConsumptionPolicy.ALLOW_DURING_BUILD_ONLY,
serverMetrics)) {
+ Semaphore semaphore =
_partitionGroupIdToConsumerCoordinatorMap.get(PARTITION_GROUP_ID).getSemaphore();
+ Assert.assertTrue(semaphore.tryAcquire());
+ segmentDataManager.getConsumerSemaphoreAcquired().set(true);
+
segmentDataManager.downloadSegmentAndReplace(crcMetadata(finalOffsetValue,
12345L));
+ Assert.assertEquals(semaphore.availablePermits(), 0);
+ verify(serverMetrics, never()).addMeteredTableValue(anyString(),
+ eq(ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD),
anyLong());
+ }
+
+ // DISALLOW_ALWAYS never lets the next segment consume during the replace,
even when this segment never acquired the
+ // semaphore.
+ serverMetrics = spy(new
ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry()));
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManagerForUpsertRevert(
+ ParallelSegmentConsumptionPolicy.DISALLOW_ALWAYS, serverMetrics)) {
+
segmentDataManager.downloadSegmentAndReplace(crcMetadata(finalOffsetValue,
12345L));
+ verify(serverMetrics, never()).addMeteredTableValue(anyString(),
+ eq(ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD),
anyLong());
+ }
+ }
+
+ private FakeRealtimeSegmentDataManager
createFakeSegmentManagerForUpsertRevert(
+ ParallelSegmentConsumptionPolicy policy, ServerMetrics serverMetrics)
+ throws Exception {
+ FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager(false, new TimeSupplier(), null, null,
+ createTableConfigWithStreamIngestion(false, policy), serverMetrics);
+ PartitionUpsertMetadataManager upsertMetadataManager =
mock(PartitionUpsertMetadataManager.class);
+
when(upsertMetadataManager.shouldRevertMetadataOnInconsistency(any())).thenReturn(true);
+
segmentDataManager.setPartitionUpsertMetadataManager(upsertMetadataManager);
+ segmentDataManager._useRealDownloadAndReplace = true;
+ return segmentDataManager;
+ }
+
private TableConfig createPauselessTableConfig()
throws Exception {
+ return createTableConfigWithStreamIngestion(true, null);
+ }
+
+ private TableConfig createTableConfigWithStreamIngestion(boolean pauseless,
+ @Nullable ParallelSegmentConsumptionPolicy policy)
+ throws Exception {
TableConfig tableConfig = createTableConfig();
StreamIngestionConfig streamIngestionConfig =
new
StreamIngestionConfig(List.of(tableConfig.getIndexingConfig().getStreamConfigs()));
- streamIngestionConfig.setPauselessConsumptionEnabled(true);
+ streamIngestionConfig.setPauselessConsumptionEnabled(pauseless);
+ streamIngestionConfig.setParallelSegmentConsumptionPolicy(policy);
IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
if (ingestionConfig == null) {
ingestionConfig = new IngestionConfig();
@@ -1500,6 +1576,11 @@ public class RealtimeSegmentDataManagerTest {
public boolean _localSegmentCrcMatchesZk = true;
// When set, isLocalSegmentCrcMatchingZk runs the real on-disk metadata
read + CRC comparison.
public boolean _useRealCrcCheck = false;
+ // When set, buildSegmentInternal releases the consumer semaphore like the
real build does when the parallel
+ // consumption policy allows consuming during build.
+ public boolean _releaseSemaphoreOnBuild = false;
+ // When set, downloadSegmentAndReplace runs the real implementation
against the mocked table data manager.
+ public boolean _useRealDownloadAndReplace = false;
private Field _streamMsgOffsetFactory;
public LinkedList<LongMsgOffset> _consumeOffsets = new LinkedList<>();
public LinkedList<SegmentCompletionProtocol.Response> _responses = new
LinkedList<>();
@@ -1569,6 +1650,15 @@ public class RealtimeSegmentDataManagerTest {
realtimeSegmentField.set(this, realtimeSegment);
}
+ /// Replaces the upsert metadata manager, which the constructor always
leaves null.
+ public void
setPartitionUpsertMetadataManager(PartitionUpsertMetadataManager
partitionUpsertMetadataManager)
+ throws Exception {
+ Field partitionUpsertMetadataManagerField =
+
RealtimeSegmentDataManager.class.getDeclaredField("_partitionUpsertMetadataManager");
+ partitionUpsertMetadataManagerField.setAccessible(true);
+ partitionUpsertMetadataManagerField.set(this,
partitionUpsertMetadataManager);
+ }
+
public boolean isStreamConsumerClosed()
throws Exception {
Field streamConsumerClosedField =
RealtimeSegmentDataManager.class.getDeclaredField("_streamConsumerClosed");
@@ -1717,6 +1807,9 @@ public class RealtimeSegmentDataManagerTest {
protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) {
terminateLoopIfNecessary();
_buildSegmentCalled = true;
+ if (_releaseSemaphoreOnBuild) {
+ closeStreamConsumerAndReleaseSemaphore();
+ }
if (_failSegmentBuild) {
try {
_segmentBuildFailedWithDeterministicError.set(this, true);
@@ -1745,9 +1838,13 @@ public class RealtimeSegmentDataManagerTest {
}
@Override
- protected void downloadSegmentAndReplace(SegmentZKMetadata metadata) {
+ protected void downloadSegmentAndReplace(SegmentZKMetadata metadata)
+ throws Exception {
terminateLoopIfNecessary();
_downloadAndReplaceCalled = true;
+ if (_useRealDownloadAndReplace) {
+ super.downloadSegmentAndReplace(metadata);
+ }
}
@Override
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
index 9c1351f6dc6..09a83c7becd 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
@@ -716,6 +716,7 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
///
/// @param oldSegment the old segment being replaced
/// @return true if metadata revert should be performed on inconsistency
+ @Override
public boolean shouldRevertMetadataOnInconsistency(IndexSegment oldSegment) {
return
ConsumingSegmentConsistencyModeListener.getInstance().getConsistencyMode()
== ConsumingSegmentConsistencyModeListener.Mode.PROTECTED
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
index 4adc57b18a2..112b2c17f55 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
@@ -74,6 +74,11 @@ public interface PartitionUpsertMetadataManager extends
Closeable {
/// Replaces the upsert metadata for the old segment with the new immutable
segment.
void replaceSegment(ImmutableSegment segment, IndexSegment oldSegment);
+ /// Returns true when replacing the given segment reverts its keys to their
previous locations.
+ default boolean shouldRevertMetadataOnInconsistency(IndexSegment oldSegment)
{
+ return false;
+ }
+
/// Removes the upsert metadata for the given segment.
void removeSegment(IndexSegment segment);
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index 2ece7905724..c77b42930a4 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -91,6 +91,7 @@ import
org.apache.pinot.spi.config.table.ingestion.ComplexTypeConfig;
import org.apache.pinot.spi.config.table.ingestion.EnrichmentConfig;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import
org.apache.pinot.spi.config.table.ingestion.ParallelSegmentConsumptionPolicy;
import
org.apache.pinot.spi.config.table.ingestion.SchemaConformingTransformerConfig;
import org.apache.pinot.spi.config.table.ingestion.SourceFieldConfig;
import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
@@ -108,6 +109,7 @@ import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Segment.AssignmentStrategy;
+import org.apache.pinot.spi.utils.ConsumingSegmentConsistencyModeListener;
import org.apache.pinot.spi.utils.DataSizeUtils;
import org.apache.pinot.spi.utils.Enablement;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
@@ -1211,6 +1213,7 @@ public final class TableConfigUtils {
validateTTLForUpsertConfig(tableConfig, schema);
validatePartialUpsertStrategies(tableConfig, schema);
+ validateConsumptionDuringUpsertRevert(tableConfig);
Preconditions.checkState(
!(PinotMd5Mode.isPinotMd5Disabled() &&
upsertConfig.getHashFunction() == HashFunction.MD5),
"Upsert hash function MD5 is disabled via '%s=true'",
CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED);
@@ -1225,6 +1228,35 @@ public final class TableConfigUtils {
}
}
+ /// Rejects consuming the next segment during a download on tables that
revert upsert metadata in PROTECTED mode,
+ /// because the next segment's snapshot would run before the revert and miss
the rows it restores.
+ @VisibleForTesting
+ static void validateConsumptionDuringUpsertRevert(TableConfig tableConfig) {
+ if (tableConfig.getTableType() != TableType.REALTIME ||
!isTableTypeInconsistentDuringConsumption(tableConfig)
+ ||
ConsumingSegmentConsistencyModeListener.getInstance().getConsistencyMode()
+ != ConsumingSegmentConsistencyModeListener.Mode.PROTECTED) {
+ return;
+ }
+ IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
+ StreamIngestionConfig streamIngestionConfig =
+ ingestionConfig != null ? ingestionConfig.getStreamIngestionConfig() :
null;
+ ParallelSegmentConsumptionPolicy policy =
+ streamIngestionConfig != null ?
streamIngestionConfig.getParallelSegmentConsumptionPolicy() : null;
+ boolean pauseless = streamIngestionConfig != null &&
streamIngestionConfig.isPauselessConsumptionEnabled();
+ // ALLOW_ALWAYS and ALLOW_DURING_DOWNLOAD_ONLY both allow it, and so does
the deprecated flag when no policy is set.
+ // ALLOW_DURING_BUILD_ONLY allows it too, because a local build that fails
or mismatches the committed CRC falls
+ // back to a download after the semaphore is released. Pauseless tables
consume during build by design and skip
+ // the CRC check, so they keep it.
+ boolean consumesDuringDownload = policy != null
+ ? policy.isAllowedDuringDownload() || (policy.isAllowedDuringBuild()
&& !pauseless)
+ :
tableConfig.getUpsertConfig().isAllowPartialUpsertConsumptionDuringCommit();
+ Preconditions.checkState(!consumesDuringDownload,
+ "%s lets the next segment consume during a segment download, but
tables with partial upsert, "
+ + "dropOutOfOrderRecord or outOfOrderRecordColumn revert upsert
metadata in PROTECTED consistency mode. "
+ + "Set parallelSegmentConsumptionPolicy to DISALLOW_ALWAYS",
+ policy != null ? "parallelSegmentConsumptionPolicy " + policy :
"allowPartialUpsertConsumptionDuringCommit");
+ }
+
/// Checks if a data type is valid for time-based comparison operations
(upsert/dedup).
/// Valid types include numeric types and types with stored data as a
numeric type:
/// e.g. TIMESTAMP which is stored as LONG internally.
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java
index d76c64d34c9..1571dfc24e3 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java
@@ -2598,6 +2598,28 @@ public class
ConcurrentMapPartitionUpsertMetadataManagerTest {
upsertMetadataManager.close();
}
+ @Test
+ public void testShouldRevertMetadataOnlyForConsumingSegmentsInProtectedMode()
+ throws IOException {
+ PartitionUpsertMetadataManager upsertMetadataManager = new
ConcurrentMapPartitionUpsertMetadataManager(
+ REALTIME_TABLE_NAME, 0,
+
_contextBuilder.setConsistencyMode(UpsertConfig.ConsistencyMode.NONE).setDropOutOfOrderRecord(true).build());
+ MutableSegment mutableSegment = mock(MutableSegment.class);
+ ImmutableSegmentImpl immutableSegment = mock(ImmutableSegmentImpl.class);
+ ConsumingSegmentConsistencyModeListener consistencyModeListener =
+ ConsumingSegmentConsistencyModeListener.getInstance();
+ try {
+
assertFalse(upsertMetadataManager.shouldRevertMetadataOnInconsistency(mutableSegment));
+
consistencyModeListener.setMode(ConsumingSegmentConsistencyModeListener.Mode.PROTECTED);
+
assertTrue(upsertMetadataManager.shouldRevertMetadataOnInconsistency(mutableSegment));
+
assertFalse(upsertMetadataManager.shouldRevertMetadataOnInconsistency(immutableSegment));
+ } finally {
+ consistencyModeListener.reset();
+ upsertMetadataManager.stop();
+ upsertMetadataManager.close();
+ }
+ }
+
@Test
public void testNoRevertForImmutableSegmentReplacement()
throws IOException {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index 97a0263b5a2..ae5af3891de 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -64,6 +64,7 @@ import
org.apache.pinot.spi.config.table.ingestion.BatchIngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.ComplexTypeConfig;
import org.apache.pinot.spi.config.table.ingestion.FilterConfig;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import
org.apache.pinot.spi.config.table.ingestion.ParallelSegmentConsumptionPolicy;
import org.apache.pinot.spi.config.table.ingestion.SourceFieldConfig;
import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.TransformConfig;
@@ -83,6 +84,7 @@ import org.apache.pinot.spi.stream.StreamMessageDecoder;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Segment.AssignmentStrategy;
+import org.apache.pinot.spi.utils.ConsumingSegmentConsistencyModeListener;
import org.apache.pinot.spi.utils.Enablement;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.PinotDataType;
@@ -2682,6 +2684,92 @@ public class TableConfigUtilsTest {
}
}
+ @Test
+ public void testRejectConsumptionDuringDownloadWhenUpsertMetadataReverts() {
+ UpsertConfig partialUpsertConfig = new
UpsertConfig(UpsertConfig.Mode.PARTIAL);
+ UpsertConfig allowDuringCommitConfig = new
UpsertConfig(UpsertConfig.Mode.PARTIAL);
+ allowDuringCommitConfig.setAllowPartialUpsertConsumptionDuringCommit(true);
+ UpsertConfig dropOutOfOrderConfig = new
UpsertConfig(UpsertConfig.Mode.FULL);
+ dropOutOfOrderConfig.setDropOutOfOrderRecord(true);
+ UpsertConfig fullUpsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
+
+ ConsumingSegmentConsistencyModeListener consistencyModeListener =
+ ConsumingSegmentConsistencyModeListener.getInstance();
+ try {
+ // Without PROTECTED mode nothing is reverted, so every policy is allowed
+ checkConsumptionDuringUpsertRevert(partialUpsertConfig,
ParallelSegmentConsumptionPolicy.ALLOW_ALWAYS, false);
+
+
consistencyModeListener.setMode(ConsumingSegmentConsistencyModeListener.Mode.PROTECTED);
+ checkConsumptionDuringUpsertRevert(partialUpsertConfig,
ParallelSegmentConsumptionPolicy.ALLOW_ALWAYS, true);
+ checkConsumptionDuringUpsertRevert(partialUpsertConfig,
+ ParallelSegmentConsumptionPolicy.ALLOW_DURING_DOWNLOAD_ONLY, true);
+ checkConsumptionDuringUpsertRevert(dropOutOfOrderConfig,
ParallelSegmentConsumptionPolicy.ALLOW_ALWAYS, true);
+ checkConsumptionDuringUpsertRevert(allowDuringCommitConfig, null, true);
+ // A local build that fails or mismatches the committed CRC falls back
to a download after releasing the semaphore
+ checkConsumptionDuringUpsertRevert(partialUpsertConfig,
ParallelSegmentConsumptionPolicy.ALLOW_DURING_BUILD_ONLY,
+ true);
+ checkConsumptionDuringUpsertRevert(partialUpsertConfig,
ParallelSegmentConsumptionPolicy.DISALLOW_ALWAYS, false);
+ checkConsumptionDuringUpsertRevert(partialUpsertConfig, null, false);
+ // An explicit policy wins over the deprecated flag, and full upsert
without out-of-order handling never reverts
+ checkConsumptionDuringUpsertRevert(allowDuringCommitConfig,
ParallelSegmentConsumptionPolicy.DISALLOW_ALWAYS,
+ false);
+ checkConsumptionDuringUpsertRevert(fullUpsertConfig,
ParallelSegmentConsumptionPolicy.ALLOW_ALWAYS, false);
+ // Pauseless tables consume during build by design, so they keep
ALLOW_DURING_BUILD_ONLY but not download overlap
+
TableConfigUtils.validateConsumptionDuringUpsertRevert(createTableConfigWithConsumptionPolicy(partialUpsertConfig,
+ ParallelSegmentConsumptionPolicy.ALLOW_DURING_BUILD_ONLY, true));
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validateConsumptionDuringUpsertRevert(
+ createTableConfigWithConsumptionPolicy(partialUpsertConfig,
ParallelSegmentConsumptionPolicy.ALLOW_ALWAYS,
+ true)));
+
+ // The check is part of the upsert validation
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .setPrimaryKeyColumns(List.of("myCol"))
+ .build();
+ TableConfig tableConfig =
+ createTableConfigWithConsumptionPolicy(dropOutOfOrderConfig,
ParallelSegmentConsumptionPolicy.ALLOW_ALWAYS);
+ IllegalStateException e = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validateUpsertAndDedupConfig(tableConfig,
schema));
+ assertTrue(e.getMessage().endsWith("Set parallelSegmentConsumptionPolicy
to DISALLOW_ALWAYS"), e.getMessage());
+ } finally {
+ consistencyModeListener.reset();
+ }
+ }
+
+ private void checkConsumptionDuringUpsertRevert(UpsertConfig upsertConfig,
ParallelSegmentConsumptionPolicy policy,
+ boolean expectRejected) {
+ TableConfig tableConfig =
createTableConfigWithConsumptionPolicy(upsertConfig, policy);
+ if (expectRejected) {
+ IllegalStateException e = expectThrows(IllegalStateException.class,
+ () ->
TableConfigUtils.validateConsumptionDuringUpsertRevert(tableConfig));
+ assertTrue(e.getMessage().endsWith("Set parallelSegmentConsumptionPolicy
to DISALLOW_ALWAYS"), e.getMessage());
+ assertTrue(e.getMessage().startsWith(
+ policy != null ? "parallelSegmentConsumptionPolicy " + policy :
"allowPartialUpsertConsumptionDuringCommit"),
+ e.getMessage());
+ } else {
+ TableConfigUtils.validateConsumptionDuringUpsertRevert(tableConfig);
+ }
+ }
+
+ private TableConfig createTableConfigWithConsumptionPolicy(UpsertConfig
upsertConfig,
+ ParallelSegmentConsumptionPolicy policy) {
+ return createTableConfigWithConsumptionPolicy(upsertConfig, policy, false);
+ }
+
+ private TableConfig createTableConfigWithConsumptionPolicy(UpsertConfig
upsertConfig,
+ ParallelSegmentConsumptionPolicy policy, boolean pauseless) {
+ StreamIngestionConfig streamIngestionConfig = new
StreamIngestionConfig(List.of(getStreamConfigs()));
+ streamIngestionConfig.setParallelSegmentConsumptionPolicy(policy);
+ streamIngestionConfig.setPauselessConsumptionEnabled(pauseless);
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setStreamIngestionConfig(streamIngestionConfig);
+ return new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
+ .setUpsertConfig(upsertConfig)
+ .setRoutingConfig(STRICT_REPLICA_ROUTING_CONFIG)
+ .setIngestionConfig(ingestionConfig)
+ .build();
+ }
+
@Test
public void testValidateUpsertConfigWithMd5Disabled() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
