This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 5771d6acea6 Defer TTL watermark bump until after segment
add/preload/replace (upsert & dedup) (#19512)
5771d6acea6 is described below
commit 5771d6acea60cd72738116835111cf7c49965373
Author: Chaitanya Deepthi <[email protected]>
AuthorDate: Fri Sep 11 18:16:15 2026 -0700
Defer TTL watermark bump until after segment add/preload/replace (upsert &
dedup) (#19512)
---
.../dedup/BasePartitionDedupMetadataManager.java | 47 ++++++++++++-------
...ConcurrentMapPartitionDedupMetadataManager.java | 5 +-
.../upsert/BasePartitionUpsertMetadataManager.java | 54 ++++++++++++++--------
...oncurrentMapPartitionUpsertMetadataManager.java | 9 ++--
...nUpsertMetadataManagerForConsistentDeletes.java | 10 ++--
...apPartitionDedupMetadataManagerWithTTLTest.java | 31 +++++++++++++
...rrentMapPartitionUpsertMetadataManagerTest.java | 33 +++++++++++++
7 files changed, 141 insertions(+), 48 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/BasePartitionDedupMetadataManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/BasePartitionDedupMetadataManager.java
index 693c085d91a..8765758da0b 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/BasePartitionDedupMetadataManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/BasePartitionDedupMetadataManager.java
@@ -167,16 +167,18 @@ public abstract class BasePartitionDedupMetadataManager
implements PartitionDedu
return;
}
try {
- if (skipSegmentOutOfTTL(segment, true)) {
- return;
- }
- try (DedupUtils.DedupRecordInfoReader dedupRecordInfoReader = new
DedupUtils.DedupRecordInfoReader(segment,
- _primaryKeyColumns, _dedupTimeColumn)) {
- Iterator<DedupRecordInfo> dedupRecordInfoIterator =
- DedupUtils.getDedupRecordInfoIterator(dedupRecordInfoReader,
segment.getSegmentMetadata().getTotalDocs());
- doPreloadSegment(segment, dedupRecordInfoIterator);
- updatePrimaryKeyGauge();
+ // Bump watermark after doPreloadSegment; a concurrent
removeExpiredPrimaryKeys sweep reading a pre-bumped
+ // watermark could expire keys this preload is about to insert.
+ if (!skipSegmentOutOfTTL(segment)) {
+ try (DedupUtils.DedupRecordInfoReader dedupRecordInfoReader = new
DedupUtils.DedupRecordInfoReader(segment,
+ _primaryKeyColumns, _dedupTimeColumn)) {
+ Iterator<DedupRecordInfo> dedupRecordInfoIterator =
+ DedupUtils.getDedupRecordInfoIterator(dedupRecordInfoReader,
segment.getSegmentMetadata().getTotalDocs());
+ doPreloadSegment(segment, dedupRecordInfoIterator);
+ updatePrimaryKeyGauge();
+ }
}
+ updateLargestSeenTime(segment);
} catch (Exception e) {
throw new RuntimeException(
String.format("Caught exception while preloading segment: %s of
table: %s in %s", segmentName,
@@ -203,9 +205,11 @@ public abstract class BasePartitionDedupMetadataManager
implements PartitionDedu
return;
}
try {
- if (!skipSegmentOutOfTTL(segment, true)) {
+ // Bump watermark after add; see preloadSegment.
+ if (!skipSegmentOutOfTTL(segment)) {
addOrReplaceSegment(null, segment);
}
+ updateLargestSeenTime(segment);
} catch (Exception e) {
throw new RuntimeException(
String.format("Caught exception while adding segment: %s of table:
%s to %s", segmentName, _tableNameWithType,
@@ -223,9 +227,11 @@ public abstract class BasePartitionDedupMetadataManager
implements PartitionDedu
return;
}
try {
- if (!skipSegmentOutOfTTL(newSegment, true)) {
+ // Bump watermark after replace; see preloadSegment.
+ if (!skipSegmentOutOfTTL(newSegment)) {
addOrReplaceSegment(oldSegment, newSegment);
}
+ updateLargestSeenTime(newSegment);
} catch (Exception e) {
throw new RuntimeException(
String.format("Caught exception while replacing segment: %s with
segment: %s of table: %s in %s",
@@ -236,16 +242,13 @@ public abstract class BasePartitionDedupMetadataManager
implements PartitionDedu
}
}
- protected boolean skipSegmentOutOfTTL(IndexSegment segment, boolean
updateWatermark) {
+ protected boolean skipSegmentOutOfTTL(IndexSegment segment) {
if (_metadataTTL <= 0) {
return false;
}
// If metadataTTL is enabled, we can skip adding dedup metadata for
segment already out of the TTL. Different
// from upsert table, there is no need to initialize things like
validDocIds bitmap for those skipped segments.
double maxDedupTime = getMaxDedupTime(segment);
- if (updateWatermark) {
- _largestSeenTime.getAndUpdate(time -> Math.max(time, maxDedupTime));
- }
if (!isOutOfMetadataTTL(maxDedupTime)) {
return false;
}
@@ -255,6 +258,18 @@ public abstract class BasePartitionDedupMetadataManager
implements PartitionDedu
return true;
}
+ protected void updateLargestSeenTime(IndexSegment segment) {
+ if (_metadataTTL > 0) {
+ updateLargestSeenTime(getMaxDedupTime(segment));
+ }
+ }
+
+ protected void updateLargestSeenTime(double dedupTime) {
+ if (_metadataTTL > 0) {
+ _largestSeenTime.getAndUpdate(time -> Math.max(time, dedupTime));
+ }
+ }
+
private void addOrReplaceSegment(@Nullable IndexSegment oldSegment,
IndexSegment newSegment)
throws IOException {
try (DedupUtils.DedupRecordInfoReader dedupRecordInfoReader = new
DedupUtils.DedupRecordInfoReader(newSegment,
@@ -281,7 +296,7 @@ public abstract class BasePartitionDedupMetadataManager
implements PartitionDedu
return;
}
try {
- if (skipSegmentOutOfTTL(segment, false)) {
+ if (skipSegmentOutOfTTL(segment)) {
return;
}
try (DedupUtils.DedupRecordInfoReader dedupRecordInfoReader = new
DedupUtils.DedupRecordInfoReader(segment,
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManager.java
index 5a1c0440bd8..bb18c7320e0 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManager.java
@@ -116,9 +116,6 @@ class ConcurrentMapPartitionDedupMetadataManager extends
BasePartitionDedupMetad
return true;
}
try {
- if (_metadataTTL > 0) {
- _largestSeenTime.getAndUpdate(time -> Math.max(time,
dedupRecordInfo.getDedupTime()));
- }
AtomicBoolean present = new AtomicBoolean(false);
_primaryKeyToSegmentAndTimeMap.compute(HashUtils.hashPrimaryKey(dedupRecordInfo.getPrimaryKey(),
_hashFunction),
(primaryKey, segmentAndTime) -> {
@@ -131,6 +128,8 @@ class ConcurrentMapPartitionDedupMetadataManager extends
BasePartitionDedupMetad
present.set(true);
return segmentAndTime;
});
+ // Bump after the map is updated to avoid racing
removeExpiredPrimaryKeys.
+ updateLargestSeenTime(dedupRecordInfo.getDedupTime());
if (!present.get()) {
updatePrimaryKeyGauge();
}
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 166ff082500..c6795111a8f 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
@@ -316,6 +316,12 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
return _metadataTTL > 0 || _deletedKeysTTL > 0;
}
+ protected void updateLargestSeenComparisonValue(double comparisonValue) {
+ if (isTTLEnabled()) {
+ _largestSeenComparisonValue.getAndUpdate(v -> Math.max(v,
comparisonValue));
+ }
+ }
+
protected double getMaxComparisonValue(IndexSegment segment) {
return ((Number)
segment.getSegmentMetadata().getColumnMetadataMap().get(_comparisonColumns.get(0))
.getMaxValue()).doubleValue();
@@ -364,12 +370,15 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
protected void doAddSegment(ImmutableSegmentImpl segment) {
String segmentName = segment.getSegmentName();
_logger.info("Adding segment: {}, current primary key count: {}",
segmentName, getNumPrimaryKeys());
- if (isTTLEnabled()) {
- double maxComparisonValue = getMaxComparisonValue(segment);
- _largestSeenComparisonValue.getAndUpdate(v -> Math.max(v,
maxComparisonValue));
- if (isOutOfMetadataTTL(maxComparisonValue) &&
skipAddSegmentOutOfTTL(segment)) {
- return;
- }
+ // Bump watermark after rows are added, otherwise a concurrent
removeExpiredPrimaryKeys can expire keys we are
+ // about to insert. Per-partition state transitions are serialized by
Helix, so concurrent doAddSegment on the
+ // same partition is not possible; the only concurrency here is with the
sweep. If addSegment throws, the
+ // watermark stays unchanged - the segment's rows are not in the map, so
recording their max would misrepresent
+ // what we have observed and let the sweep expire other keys against a
phantom watermark.
+ double maxComparisonValue = isTTLEnabled() ?
getMaxComparisonValue(segment) : TTL_WATERMARK_NOT_SET;
+ if (isTTLEnabled() && isOutOfMetadataTTL(maxComparisonValue) &&
skipAddSegmentOutOfTTL(segment)) {
+ updateLargestSeenComparisonValue(maxComparisonValue);
+ return;
}
long startTimeMs = System.currentTimeMillis();
if (!_enableSnapshot) {
@@ -381,9 +390,14 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
UpsertUtils.getRecordInfoIterator(recordInfoReader,
segment.getSegmentMetadata().getTotalDocs());
addSegment(segment, null, null, recordInfoIterator);
} catch (Exception e) {
+ // On failure, do not bump the watermark: the segment's rows are not in
the map, and bumping would let the
+ // sweep expire other keys against a phantom watermark. Helix retries
the transition and re-bumps.
throw new RuntimeException(
String.format("Caught exception while adding segment: %s, table:
%s", segmentName, _tableNameWithType), e);
}
+ if (isTTLEnabled()) {
+ updateLargestSeenComparisonValue(maxComparisonValue);
+ }
// Update metrics
long numPrimaryKeys = getNumPrimaryKeys();
@@ -442,12 +456,12 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
segment.enableUpsert(this, new ThreadSafeMutableRoaringBitmap(),
queryableDocIds);
return;
}
- if (isTTLEnabled()) {
- double maxComparisonValue = getMaxComparisonValue(segment);
- _largestSeenComparisonValue.getAndUpdate(v -> Math.max(v,
maxComparisonValue));
- if (isOutOfMetadataTTL(maxComparisonValue) &&
skipPreloadSegmentOutOfTTL(segment, validDocIds)) {
- return;
- }
+ // Bump watermark after rows are added; see doAddSegment.
+ double maxComparisonValue = isTTLEnabled() ?
getMaxComparisonValue(segment) : TTL_WATERMARK_NOT_SET;
+ if (isTTLEnabled() && isOutOfMetadataTTL(maxComparisonValue)
+ && skipPreloadSegmentOutOfTTL(segment, validDocIds)) {
+ updateLargestSeenComparisonValue(maxComparisonValue);
+ return;
}
try (UpsertUtils.RecordInfoReader recordInfoReader = new
UpsertUtils.RecordInfoReader(segment, _primaryKeyColumns,
_comparisonColumns, _deleteRecordColumn)) {
@@ -457,6 +471,9 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
String.format("Caught exception while preloading segment: %s, table:
%s", segmentName, _tableNameWithType),
e);
}
+ if (isTTLEnabled()) {
+ updateLargestSeenComparisonValue(maxComparisonValue);
+ }
// Update metrics
long numPrimaryKeys = getNumPrimaryKeys();
@@ -596,12 +613,10 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
replaceSegment(segment, null, null, null, oldSegment);
return;
}
- if (isTTLEnabled()) {
- double maxComparisonValue = getMaxComparisonValue(segment);
- _largestSeenComparisonValue.getAndUpdate(v -> Math.max(v,
maxComparisonValue));
- // Segment might be uploaded directly to the table to replace an old
segment. So update the TTL watermark but
- // we can't skip segment even if it's out of TTL as its validDocIds
bitmap is not updated yet.
- }
+ // Bump watermark after replaceSegment; see doAddSegment. A segment may be
uploaded directly to the table to
+ // replace an old one; the incoming validDocIds bitmap is not populated
yet, so we cannot skip even if the
+ // segment is out of TTL.
+ double maxComparisonValue = isTTLEnabled() ?
getMaxComparisonValue(segment) : TTL_WATERMARK_NOT_SET;
try (UpsertUtils.RecordInfoReader recordInfoReader = new
UpsertUtils.RecordInfoReader(segment, _primaryKeyColumns,
_comparisonColumns, _deleteRecordColumn)) {
// Reload-only fast path for an upsert + TTL table. The incoming segment
carries a validDocIds snapshot ONLY when
@@ -633,6 +648,9 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
String.format("Caught exception while replacing segment: %s, table:
%s, message: %s", segmentName,
_tableNameWithType, e.getMessage()), e);
}
+ if (isTTLEnabled()) {
+ updateLargestSeenComparisonValue(maxComparisonValue);
+ }
// Update metrics
long numPrimaryKeys = getNumPrimaryKeys();
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java
index 57559b02069..f10ce13eead 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java
@@ -369,11 +369,6 @@ public class ConcurrentMapPartitionUpsertMetadataManager
extends BasePartitionUp
int newDocId = recordInfo.getDocId();
Comparable newComparisonValue = recordInfo.getComparisonValue();
- // When TTL is enabled, update largestSeenComparisonValue when adding new
record
- if (isTTLEnabled()) {
- double comparisonValue = ((Number) newComparisonValue).doubleValue();
- _largestSeenComparisonValue.getAndUpdate(v -> Math.max(v,
comparisonValue));
- }
_primaryKeyToRecordLocationMap.compute(HashUtils.hashPrimaryKey(recordInfo.getPrimaryKey(),
_hashFunction),
(primaryKey, currentRecordLocation) -> {
if (currentRecordLocation != null) {
@@ -409,6 +404,10 @@ public class ConcurrentMapPartitionUpsertMetadataManager
extends BasePartitionUp
return new RecordLocation(segment, newDocId, newComparisonValue);
}
});
+ // Bump after the record is installed to avoid racing
removeExpiredPrimaryKeys.
+ if (isTTLEnabled()) {
+ updateLargestSeenComparisonValue(((Number)
newComparisonValue).doubleValue());
+ }
updatePrimaryKeyGauge();
return !isOutOfOrderRecord.get();
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java
index 32352f7a3b0..c54dac31056 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java
@@ -490,12 +490,6 @@ public class
ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes
int newDocId = recordInfo.getDocId();
Comparable newComparisonValue = recordInfo.getComparisonValue();
- // When TTL is enabled, update largestSeenComparisonValue when adding new
record
- if (_deletedKeysTTL > 0) {
- double comparisonValue = ((Number) newComparisonValue).doubleValue();
- _largestSeenComparisonValue.getAndUpdate(v -> Math.max(v,
comparisonValue));
- }
-
_primaryKeyToRecordLocationMap.compute(HashUtils.hashPrimaryKey(recordInfo.getPrimaryKey(),
_hashFunction),
(primaryKey, currentRecordLocation) -> {
if (currentRecordLocation != null) {
@@ -540,6 +534,10 @@ public class
ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes
return new RecordLocation(segment, newDocId, newComparisonValue,
1);
}
});
+ // Bump after the record is installed; see
ConcurrentMapPartitionUpsertMetadataManager#doAddRecord.
+ if (_deletedKeysTTL > 0) {
+ updateLargestSeenComparisonValue(((Number)
newComparisonValue).doubleValue());
+ }
updatePrimaryKeyGauge();
return !isOutOfOrderRecord.get();
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManagerWithTTLTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManagerWithTTLTest.java
index 678d0064af6..3d9ff6ee88e 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManagerWithTTLTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManagerWithTTLTest.java
@@ -437,6 +437,37 @@ public class
ConcurrentMapPartitionDedupMetadataManagerWithTTLTest {
verifyAddSegmentAfterStop(HashFunction.MURMUR3);
}
+ // skipSegmentOutOfTTL must not bump the watermark: caller does that after
the rows are added, so a concurrent
+ // sweep cannot expire keys the in-flight add is about to insert.
+ @Test
+ public void testSkipSegmentOutOfTTLDoesNotBumpWatermark()
+ throws IOException {
+ _dedupContextBuilder.setHashFunction(HashFunction.NONE);
+ ConcurrentMapPartitionDedupMetadataManager metadataManager =
+ new
ConcurrentMapPartitionDedupMetadataManager(DedupTestUtils.REALTIME_TABLE_NAME,
0,
+ _dedupContextBuilder.build());
+
+ metadataManager._largestSeenTime.set(5000);
+
+ IndexSegment segment = DedupTestUtils.mockSegment(1, 10);
+ SegmentMetadataImpl segmentMetadata = mock(SegmentMetadataImpl.class);
+ ColumnMetadata columnMetadata = mock(ColumnMetadata.class);
+ when(segmentMetadata.getColumnMetadataMap()).thenReturn(new TreeMap<>() {{
+ this.put(DEDUP_TIME_COLUMN_NAME, columnMetadata);
+ }});
+ doReturn(10000.0).when(columnMetadata).getMaxValue();
+ when(segment.getSegmentMetadata()).thenReturn(segmentMetadata);
+
+ assertFalse(metadataManager.skipSegmentOutOfTTL(segment));
+ assertEquals(metadataManager._largestSeenTime.get(), 5000.0);
+
+ metadataManager.updateLargestSeenTime(segment);
+ assertEquals(metadataManager._largestSeenTime.get(), 10000.0);
+
+ metadataManager.stop();
+ metadataManager.close();
+ }
+
private void verifyAddSegmentAfterStop(HashFunction hashFunction) {
_dedupContextBuilder.setHashFunction(hashFunction);
ConcurrentMapPartitionDedupMetadataManager metadataManager =
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 a3dcfb12821..06b97a2f1dc 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
@@ -265,6 +265,39 @@ public class
ConcurrentMapPartitionUpsertMetadataManagerTest {
upsertMetadataManager.close();
}
+ // Watermark must not advance before addSegment inserts rows, or a
concurrent sweep can drop the incoming keys.
+ @Test
+ public void testDoAddSegmentBumpsWatermarkAfterRowInsert()
+ throws Exception {
+ _contextBuilder.setEnableSnapshot(true).setMetadataTTL(30);
+ double[] watermarkDuringAdd = {Double.NaN};
+ ConcurrentMapPartitionUpsertMetadataManager upsertMetadataManager =
+ new ConcurrentMapPartitionUpsertMetadataManager(REALTIME_TABLE_NAME,
0, _contextBuilder.build()) {
+ @Override
+ public void addSegment(ImmutableSegmentImpl segment, @Nullable
ThreadSafeMutableRoaringBitmap validDocIds,
+ @Nullable ThreadSafeMutableRoaringBitmap queryableDocIds,
Iterator<RecordInfo> recordInfoIterator) {
+ watermarkDuringAdd[0] = getWatermark();
+ super.addSegment(segment, validDocIds, queryableDocIds,
recordInfoIterator);
+ }
+ };
+
+ ThreadSafeMutableRoaringBitmap seedValidDocIds = new
ThreadSafeMutableRoaringBitmap();
+ MutableSegment seedSegment = mockMutableSegment(1, seedValidDocIds, null);
+ upsertMetadataManager.addRecord(seedSegment, new
RecordInfo(makePrimaryKey(999), 0, 100, false));
+ assertEquals(upsertMetadataManager.getWatermark(), 100.0);
+
+ ThreadSafeMutableRoaringBitmap validDocIds = new
ThreadSafeMutableRoaringBitmap();
+ ImmutableSegmentImpl segment =
+ createRealSegment("deferred_bump_segment", new int[]{1, 2, 3}, new
int[]{150, 180, 200}, validDocIds);
+ upsertMetadataManager.addSegment(segment);
+
+ assertEquals(watermarkDuringAdd[0], 100.0);
+ assertEquals(upsertMetadataManager.getWatermark(), 200.0);
+
+ upsertMetadataManager.stop();
+ upsertMetadataManager.close();
+ }
+
@Test
public void testManageWatermark()
throws IOException {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
