This is an automated email from the ASF dual-hosted git repository.
tarun11Mavani 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 684b7a4aa5f Add metric for local realtime segment builds with CRC
mismatch (#19450)
684b7a4aa5f is described below
commit 684b7a4aa5f66173ffdfac058713e75adaec123f
Author: Prabhat Singh <[email protected]>
AuthorDate: Thu Nov 5 17:52:05 2026 +0530
Add metric for local realtime segment builds with CRC mismatch (#19450)
* Add metric for CRC-driven segment replacements
Expose table-level observability for replicas that replace a segment after
a confirmed CRC mismatch. Count only successful replacements so failed
downloads and CRC read errors do not inflate the signal.
Co-authored-by: Cursor <[email protected]>
# Conflicts:
#
pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
#
pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
* Count CRC-driven replacements on the reloadSegment path
reloadSegment can download and replace a segment after a CRC mismatch
without going through replaceSegmentIfCrcMismatch, so the metric was
missing those successful replacements. Same-CRC force downloads and
failed reloads remain excluded.
Co-authored-by: Cursor <[email protected]>
# Conflicts:
#
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
* Count confirmed CRC mismatches on locally built realtime segments
Rename the meter to SEGMENT_BUILD_WITH_CRC_MISMATCH and emit it when a
local build diverges from the committed CRC, even if download fallback
fails. Drop BuildSegmentResult and BaseTableDataManager increments so
the metric stays scoped to the local-build path.
Co-authored-by: Cursor <[email protected]>
---------
Co-authored-by: Cursor <[email protected]>
---
.../apache/pinot/common/metrics/ServerMeter.java | 1 +
.../realtime/RealtimeSegmentDataManager.java | 41 +++--
.../data/manager/BaseTableDataManagerTest.java | 18 ++
.../realtime/RealtimeSegmentDataManagerTest.java | 181 +++++++++++++++++----
4 files changed, 195 insertions(+), 46 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 a02882ce34b..ce15566628d 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
@@ -103,6 +103,7 @@ public enum ServerMeter implements AbstractMetrics.Meter {
SEGMENT_DOWNLOAD_FAILURES("segments", false),
SEGMENT_DOWNLOAD_FROM_REMOTE_FAILURES("segments", false),
SEGMENT_DOWNLOAD_FROM_PEERS_FAILURES("segments", false),
+ SEGMENT_BUILD_WITH_CRC_MISMATCH("segments", false),
SEGMENT_BUILD_FAILURE("segments", false),
SEGMENT_UPLOAD_FAILURE("segments", false),
SEGMENT_UPLOAD_SUCCESS("segments", false),
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 df626b8e6c0..df9b875fed3 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
@@ -40,6 +40,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.function.BooleanSupplier;
import javax.annotation.Nullable;
+import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.Utils;
@@ -1410,12 +1411,15 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
///
/// A locally-built segment can diverge in CRC from the one the winning
replica committed (e.g. different docId
/// ordering); swapping in a divergent copy corrupts upsert metadata and
leaves replicas inconsistent otherwise. So
- /// when `committedSegmentZKMetadata` carries a CRC, a mismatch discards the
local build (returns `false`) and the
+ /// when `committedSegmentZKMetadata` carries a CRC, a mismatch discards the
local build and the
/// caller downloads the committed segment. Skipped for pauseless tables and
segments whose CRC is not yet set (the
/// COMMITTING window), so we never download a not-yet-uploaded segment (see
PR #17885).
///
+ /// Confirmed CRC mismatches are counted via {@link
ServerMeter#SEGMENT_BUILD_WITH_CRC_MISMATCH} regardless of
+ /// whether the caller's download fallback succeeds. Build failures and
CRC-read errors are not counted.
+ ///
/// @param committedSegmentZKMetadata committed ZK metadata carrying the
CRC, or `null` to skip the check
- /// @return `true` if built and replaced locally; `false` if the build
failed or was rejected on a CRC mismatch
+ /// @return `true` if the consuming segment was replaced with the local
copy; `false` if the caller should download
protected boolean buildSegmentAndReplace(@Nullable SegmentZKMetadata
committedSegmentZKMetadata)
throws Exception {
SegmentBuildDescriptor descriptor;
@@ -1429,10 +1433,19 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
}
boolean crcCheckEnabled = committedSegmentZKMetadata != null &&
committedSegmentZKMetadata.getCrc() >= 0
&& !PauselessConsumptionUtils.isPauselessEnabled(_tableConfig);
- if (crcCheckEnabled &&
!isLocalSegmentCrcMatchingZk(committedSegmentZKMetadata)) {
- _segmentLogger.warn("Locally-built segment: {} CRC does not match
committed CRC: {} in zk. "
- + "Skipping local build to replace", _segmentNameStr,
committedSegmentZKMetadata.getCrc());
- return false;
+ if (crcCheckEnabled) {
+ try {
+ if (!isLocalSegmentCrcMatchingZk(committedSegmentZKMetadata)) {
+ _segmentLogger.warn("Locally-built segment: {} CRC does not match
committed CRC: {} in zk. "
+ + "Skipping local build to replace", _segmentNameStr,
committedSegmentZKMetadata.getCrc());
+ _serverMetrics.addMeteredTableValue(_tableNameWithType,
ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH, 1);
+ return false;
+ }
+ } catch (Exception e) {
+ _segmentLogger.warn("Failed to read CRC of locally-built segment: {};
treating as a build failure to download "
+ + "the committed segment", _segmentNameStr, e);
+ return false;
+ }
}
// On a CRC match use the committed metadata; otherwise the
construction-time metadata.
_realtimeTableDataManager.replaceConsumingSegment(_segmentNameStr,
@@ -1440,19 +1453,13 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
return true;
}
- /// Whether the just-built local segment's CRC matches the committed CRC in
ZK. Returns `false` if the local metadata
- /// cannot be read, so the caller downloads the committed segment rather
than failing the transition.
+ /// Whether the just-built local segment's CRC matches the committed CRC in
ZK.
@VisibleForTesting
- protected boolean isLocalSegmentCrcMatchingZk(SegmentZKMetadata
committedSegmentZKMetadata) {
+ protected boolean isLocalSegmentCrcMatchingZk(SegmentZKMetadata
committedSegmentZKMetadata)
+ throws IOException, ConfigurationException {
File indexDir = new File(_resourceDataDir, _segmentNameStr);
- try {
- SegmentMetadataImpl localMetadata = new SegmentMetadataImpl(indexDir);
- return BaseTableDataManager.hasSameCRC(committedSegmentZKMetadata,
localMetadata);
- } catch (Exception e) {
- _segmentLogger.warn("Failed to read CRC of locally-built segment: {};
treating as CRC mismatch to download the "
- + "committed segment", _segmentNameStr, e);
- return false;
- }
+ SegmentMetadataImpl localMetadata = new SegmentMetadataImpl(indexDir);
+ return BaseTableDataManager.hasSameCRC(committedSegmentZKMetadata,
localMetadata);
}
/// Closes the stream consumer so that no more data can be consumed into
this segment. Does NOT release the consumer
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
index 45136d68169..b5ce0eca72a 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
@@ -939,6 +939,24 @@ public class BaseTableDataManagerTest {
assertFalse(tableDataManager.getSegmentDataDir(segmentName).exists());
}
+ @Test
+ public void testReplaceSegmentSkippedWhenCrcCheckDisabled()
+ throws Exception {
+ SegmentZKMetadata zkMetadata = createRawSegment(SegmentVersion.v3, 5);
+ InstanceDataManagerConfig instanceDataManagerConfig =
createDefaultInstanceDataManagerConfig();
+
when(instanceDataManagerConfig.shouldCheckCRCOnSegmentLoad()).thenReturn(false);
+
+ BaseTableDataManager tableDataManager =
createTableManager(instanceDataManagerConfig);
+ File indexDir = createSegment(SegmentVersion.v3, 3);
+ tableDataManager.addNewOnlineSegment(zkMetadata, new IndexLoadingConfig());
+ assertEquals(new SegmentMetadataImpl(indexDir).getTotalDocs(), 3);
+
+ ImmutableSegmentDataManager segmentDataManager =
createImmutableSegmentDataManager(SEGMENT_NAME, 0);
+ tableDataManager.replaceSegmentIfCrcMismatch(segmentDataManager,
zkMetadata, new IndexLoadingConfig());
+ // CRC check disabled: local copy is kept even though CRCs differ.
+ assertEquals(new SegmentMetadataImpl(indexDir).getTotalDocs(), 3);
+ }
+
@Test
public void testReplaceSegmentNewDataWithAsyncSegmentRefresh()
throws Exception {
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 64f1b96efb9..f3e3d3dab3a 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
@@ -37,6 +37,7 @@ import java.util.concurrent.locks.Lock;
import java.util.function.BooleanSupplier;
import java.util.function.Supplier;
import javax.annotation.Nullable;
+import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.io.FileUtils;
import org.apache.helix.HelixManager;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
@@ -102,6 +103,9 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
// TODO Re-write this test using the stream abstraction
@@ -778,7 +782,9 @@ public class RealtimeSegmentDataManagerTest {
}
// Test downloadAndReplace is called after buildAndReplace fails.
- try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ ServerMetrics buildFailureServerMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
buildFailureServerMetrics)) {
segmentDataManager._failSegmentBuildAndReplace = true;
segmentDataManager._stopWaitTimeMs = 0;
segmentDataManager._state.set(segmentDataManager,
RealtimeSegmentDataManager.State.HOLDING);
@@ -786,6 +792,8 @@ public class RealtimeSegmentDataManagerTest {
segmentDataManager.goOnlineFromConsuming(metadata);
Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
Assert.assertTrue(segmentDataManager._buildAndReplaceCalled);
+ verify(buildFailureServerMetrics, never())
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
}
}
@@ -892,11 +900,15 @@ public class RealtimeSegmentDataManagerTest {
long finalOffsetValue = START_OFFSET_VALUE + 600;
// Upsert + committed CRC set + local CRC mismatch -> discard local build,
download the committed segment.
- try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ ServerMetrics crcMismatchServerMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
crcMismatchServerMetrics)) {
when(segmentDataManager._tableDataManager.isUpsertEnabled()).thenReturn(true);
runGoOnlineForCrcGuard(segmentDataManager, crcMetadata(finalOffsetValue,
12345L), false, finalOffsetValue);
- Assert.assertTrue(segmentDataManager._buildAndReplaceCalled);
- Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
+ assertTrue(segmentDataManager._buildAndReplaceCalled);
+ assertTrue(segmentDataManager._downloadAndReplaceCalled);
+ verify(crcMismatchServerMetrics)
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
}
// Non-upsert + committed CRC set + local CRC mismatch -> discard local
build, download the committed segment.
@@ -908,51 +920,175 @@ public class RealtimeSegmentDataManagerTest {
}
// Upsert + committed CRC set + local CRC match -> build locally, no
download, swap in committed metadata.
- try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ ServerMetrics matchingUpsertServerMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
matchingUpsertServerMetrics)) {
when(segmentDataManager._tableDataManager.isUpsertEnabled()).thenReturn(true);
SegmentZKMetadata committedMetadata = crcMetadata(finalOffsetValue,
12345L);
runGoOnlineForCrcGuard(segmentDataManager, committedMetadata, true,
finalOffsetValue);
Assert.assertTrue(segmentDataManager._buildAndReplaceCalled);
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
+ verify(matchingUpsertServerMetrics, never())
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
verify(segmentDataManager._tableDataManager)
.replaceConsumingSegment(eq(SEGMENT_NAME_STR),
same(committedMetadata));
}
// Non-upsert + committed CRC set + local CRC match -> build locally, no
download, swap in committed metadata.
- try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ ServerMetrics matchingNonUpsertServerMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
matchingNonUpsertServerMetrics)) {
when(segmentDataManager._tableDataManager.isUpsertEnabled()).thenReturn(false);
SegmentZKMetadata committedMetadata = crcMetadata(finalOffsetValue,
12345L);
runGoOnlineForCrcGuard(segmentDataManager, committedMetadata, true,
finalOffsetValue);
Assert.assertTrue(segmentDataManager._buildAndReplaceCalled);
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
+ verify(matchingNonUpsertServerMetrics, never())
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
verify(segmentDataManager._tableDataManager)
.replaceConsumingSegment(eq(SEGMENT_NAME_STR),
same(committedMetadata));
}
// Committed CRC unset (-1) -> guard skipped, build locally with
construction-time metadata.
- try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ ServerMetrics unsetCrcServerMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
unsetCrcServerMetrics)) {
when(segmentDataManager._tableDataManager.isUpsertEnabled()).thenReturn(false);
SegmentZKMetadata committedMetadata = crcMetadata(finalOffsetValue, -1L);
runGoOnlineForCrcGuard(segmentDataManager, committedMetadata, false,
finalOffsetValue);
Assert.assertTrue(segmentDataManager._buildAndReplaceCalled);
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
+ verify(unsetCrcServerMetrics, never())
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
verify(segmentDataManager._tableDataManager)
.replaceConsumingSegment(eq(SEGMENT_NAME_STR), argThat(m -> m !=
committedMetadata));
}
// Pauseless table -> guard skipped, build locally with construction-time
metadata.
+ ServerMetrics pauselessServerMetrics = mock(ServerMetrics.class);
try (FakeRealtimeSegmentDataManager segmentDataManager =
- createFakeSegmentManager(false, new TimeSupplier(), null, null,
createPauselessTableConfig())) {
+ createFakeSegmentManager(false, new TimeSupplier(), null, null,
createPauselessTableConfig(),
+ pauselessServerMetrics)) {
when(segmentDataManager._tableDataManager.isUpsertEnabled()).thenReturn(false);
SegmentZKMetadata committedMetadata = crcMetadata(finalOffsetValue,
12345L);
runGoOnlineForCrcGuard(segmentDataManager, committedMetadata, false,
finalOffsetValue);
Assert.assertTrue(segmentDataManager._buildAndReplaceCalled);
Assert.assertFalse(segmentDataManager._downloadAndReplaceCalled);
+ verify(pauselessServerMetrics, never())
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
verify(segmentDataManager._tableDataManager)
.replaceConsumingSegment(eq(SEGMENT_NAME_STR), argThat(m -> m !=
committedMetadata));
}
}
+ @Test
+ public void testCrcBuildMismatchMetricForCatchup()
+ throws Exception {
+ long finalOffsetValue = START_OFFSET_VALUE + 600;
+ LongMsgOffset finalOffset = new LongMsgOffset(finalOffsetValue);
+ SegmentZKMetadata metadata = crcMetadata(finalOffsetValue, 12345L);
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
serverMetrics)) {
+ segmentDataManager._useRealBuildAndReplace = true;
+ segmentDataManager._localSegmentCrcMatchesZk = false;
+ segmentDataManager.getConsumerSemaphoreAcquired().set(true);
+ segmentDataManager._stopWaitTimeMs = 0;
+ segmentDataManager._state.set(segmentDataManager,
RealtimeSegmentDataManager.State.CATCHING_UP);
+ segmentDataManager._consumeOffsets.add(finalOffset);
+
+ segmentDataManager.goOnlineFromConsuming(metadata);
+
+ assertTrue(segmentDataManager._buildAndReplaceCalled);
+ assertTrue(segmentDataManager._downloadAndReplaceCalled);
+ verify(serverMetrics)
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
+ }
+ }
+
+ @Test
+ public void testCrcBuildMismatchMetricNotEmittedForCatchupBuildFailure()
+ throws Exception {
+ long finalOffsetValue = START_OFFSET_VALUE + 600;
+ SegmentZKMetadata metadata = crcMetadata(finalOffsetValue, 12345L);
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
serverMetrics)) {
+ segmentDataManager._failSegmentBuildAndReplace = true;
+ segmentDataManager.getConsumerSemaphoreAcquired().set(true);
+ segmentDataManager._stopWaitTimeMs = 0;
+ segmentDataManager._state.set(segmentDataManager,
RealtimeSegmentDataManager.State.CATCHING_UP);
+ segmentDataManager._consumeOffsets.add(new
LongMsgOffset(finalOffsetValue));
+
+ segmentDataManager.goOnlineFromConsuming(metadata);
+
+ assertTrue(segmentDataManager._buildAndReplaceCalled);
+ assertTrue(segmentDataManager._downloadAndReplaceCalled);
+ verify(serverMetrics, never())
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
+ }
+ }
+
+ @Test
+ public void testCrcBuildMismatchMetricNotEmittedForCrcReadError()
+ throws Exception {
+ long finalOffsetValue = START_OFFSET_VALUE + 600;
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
serverMetrics)) {
+ segmentDataManager._useRealCrcCheck = true;
+ FileUtils.deleteQuietly(new File(new File(TEMP_DIR,
REALTIME_TABLE_NAME), SEGMENT_NAME_STR));
+ runGoOnlineForCrcGuard(segmentDataManager, crcMetadata(finalOffsetValue,
12345L), false, finalOffsetValue);
+
+ assertTrue(segmentDataManager._buildAndReplaceCalled);
+ assertTrue(segmentDataManager._downloadAndReplaceCalled);
+ verify(serverMetrics, never())
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
+ }
+ }
+
+ @Test
+ public void testCrcBuildMismatchMetricEmittedWhenMatchedOffsetDownloadFails()
+ throws Exception {
+ long finalOffsetValue = START_OFFSET_VALUE + 600;
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
serverMetrics)) {
+ segmentDataManager._failDownloadAndReplace = true;
+
+ assertThrows(IOException.class, () ->
+ runGoOnlineForCrcGuard(segmentDataManager,
crcMetadata(finalOffsetValue, 12345L), false, finalOffsetValue));
+
+ // Confirmed mismatch is counted even when the download fallback fails.
+ verify(serverMetrics)
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
+ }
+ }
+
+ @Test
+ public void testCrcBuildMismatchMetricEmittedWhenCatchupDownloadFails()
+ throws Exception {
+ long finalOffsetValue = START_OFFSET_VALUE + 600;
+ SegmentZKMetadata metadata = crcMetadata(finalOffsetValue, 12345L);
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(false, new TimeSupplier(), null, null, null,
serverMetrics)) {
+ segmentDataManager._useRealBuildAndReplace = true;
+ segmentDataManager._localSegmentCrcMatchesZk = false;
+ segmentDataManager._failDownloadAndReplace = true;
+ segmentDataManager.getConsumerSemaphoreAcquired().set(true);
+ segmentDataManager._stopWaitTimeMs = 0;
+ segmentDataManager._state.set(segmentDataManager,
RealtimeSegmentDataManager.State.CATCHING_UP);
+ segmentDataManager._consumeOffsets.add(new
LongMsgOffset(finalOffsetValue));
+
+ assertThrows(IOException.class, () ->
segmentDataManager.goOnlineFromConsuming(metadata));
+
+ // Confirmed mismatch is counted even when the download fallback fails.
+ verify(serverMetrics)
+ .addMeteredTableValue(anyString(),
eq(ServerMeter.SEGMENT_BUILD_WITH_CRC_MISMATCH), eq(1L));
+ }
+ }
+
private void runGoOnlineForCrcGuard(FakeRealtimeSegmentDataManager
segmentDataManager, SegmentZKMetadata metadata,
boolean localCrcMatchesZk, long currentOffsetValue)
throws Exception {
@@ -1082,26 +1218,11 @@ public class RealtimeSegmentDataManagerTest {
SegmentZKMetadata matchingMetadata = new
SegmentZKMetadata(SEGMENT_NAME_STR);
matchingMetadata.setCrc(localCrc);
-
Assert.assertTrue(segmentDataManager.isLocalSegmentCrcMatchingZk(matchingMetadata));
+
assertTrue(segmentDataManager.isLocalSegmentCrcMatchingZk(matchingMetadata));
SegmentZKMetadata mismatchingMetadata = new
SegmentZKMetadata(SEGMENT_NAME_STR);
mismatchingMetadata.setCrc(localCrc + 1);
-
Assert.assertFalse(segmentDataManager.isLocalSegmentCrcMatchingZk(mismatchingMetadata));
- }
- }
-
- // When the locally-built segment cannot be read (e.g. missing/corrupt), the
CRC check must report a mismatch so the
- // ONLINE transition downloads the committed segment instead of throwing.
- @Test
- public void
testIsLocalSegmentCrcMatchingZkTreatsUnreadableSegmentAsMismatch()
- throws Exception {
- try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
- segmentDataManager._useRealCrcCheck = true;
- // No segment was built under the resource dir, so the metadata read
fails.
- FileUtils.deleteQuietly(new File(new File(TEMP_DIR,
REALTIME_TABLE_NAME), SEGMENT_NAME_STR));
- SegmentZKMetadata metadata = new SegmentZKMetadata(SEGMENT_NAME_STR);
- metadata.setCrc(12345L);
-
Assert.assertFalse(segmentDataManager.isLocalSegmentCrcMatchingZk(metadata));
+
assertFalse(segmentDataManager.isLocalSegmentCrcMatchingZk(mismatchingMetadata));
}
}
@@ -1671,11 +1792,8 @@ public class RealtimeSegmentDataManagerTest {
public Field _stopReasonCode;
public Field _segmentBuildFailedWithDeterministicError;
public boolean _failSegmentBuildAndReplace = false;
- // When set, buildSegmentAndReplace runs the real implementation
(including the upsert CRC guard) instead of being
- // short-circuited, and isLocalSegmentCrcMatchingZk returns
_localSegmentCrcMatchesZk.
public boolean _useRealBuildAndReplace = false;
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.
@@ -1689,6 +1807,7 @@ public class RealtimeSegmentDataManagerTest {
public boolean _buildSegmentCalled = false;
public boolean _failSegmentBuild = false;
public boolean _buildAndReplaceCalled = false;
+ public boolean _failDownloadAndReplace = false;
public int _stopWaitTimeMs = 100;
private boolean _downloadAndReplaceCalled = false;
private boolean _notifySegmentBuildFailedWithDeterministicErrorCalled =
false;
@@ -1897,7 +2016,8 @@ public class RealtimeSegmentDataManagerTest {
}
@Override
- protected boolean isLocalSegmentCrcMatchingZk(SegmentZKMetadata
committedSegmentZKMetadata) {
+ protected boolean isLocalSegmentCrcMatchingZk(SegmentZKMetadata
committedSegmentZKMetadata)
+ throws IOException, ConfigurationException {
if (_useRealCrcCheck) {
return super.isLocalSegmentCrcMatchingZk(committedSegmentZKMetadata);
}
@@ -1943,6 +2063,9 @@ public class RealtimeSegmentDataManagerTest {
protected void downloadSegmentAndReplace(SegmentZKMetadata metadata)
throws Exception {
terminateLoopIfNecessary();
+ if (_failDownloadAndReplace) {
+ throw new IOException("download failed");
+ }
_downloadAndReplaceCalled = true;
if (_useRealDownloadAndReplace) {
super.downloadSegmentAndReplace(metadata);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
