From c57c3b561ad73d3d8f11d50719ea1d3daa321eb1 Mon Sep 17 00:00:00 2001 From: Dhananjay Sawner Date: Mon, 22 Jun 2026 17:28:59 +0530 Subject: [PATCH 1/5] Add create-validation time to stream provisioning metric recordStreamProvisioningTime now adds the resource provider's create-validation time (datastream metadata system.createValidationTime.ms) on top of (now - system.creation.ms), so the stream provisioning histogram and the within/outside-SLA counters reflect validation plus creation time. Absent or malformed values are treated as 0, keeping behavior identical for streams created before this property existed. Also raise the default provisioning SLA threshold from 8 to 10 minutes. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../common/DatastreamMetadataConstants.java | 7 +++++++ .../datastream/server/Coordinator.java | 21 ++++++++++++++++++- .../datastream/server/CoordinatorConfig.java | 2 +- .../server/TestCoordinatorConfig.java | 14 +++++++++++++ 4 files changed, 42 insertions(+), 2 deletions(-) diff --git a/datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamMetadataConstants.java b/datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamMetadataConstants.java index efcc226d8..87d10a81e 100644 --- a/datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamMetadataConstants.java +++ b/datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamMetadataConstants.java @@ -89,6 +89,13 @@ public class DatastreamMetadataConstants { */ public static final String CREATION_MS = "system.creation.ms"; + /** + * Time spent running create validation in the resource provider, in milliseconds. Added to the + * stream provisioning time alongside {@link #CREATION_MS}. Absent for streams created before this + * property existed, in which case it is treated as zero. + */ + public static final String CREATE_VALIDATION_TIME_MS = "system.createValidationTime.ms"; + /** * Position at which the ingestion should start for the datastream. */ diff --git a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java index 0e82b125e..6c952a4d8 100644 --- a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java +++ b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java @@ -86,6 +86,7 @@ import com.linkedin.datastream.server.providers.ZookeeperCheckpointProvider; import com.linkedin.datastream.server.zk.ZkAdapter; +import static com.linkedin.datastream.common.DatastreamMetadataConstants.CREATE_VALIDATION_TIME_MS; import static com.linkedin.datastream.common.DatastreamMetadataConstants.CREATION_MS; import static com.linkedin.datastream.common.DatastreamMetadataConstants.SYSTEM_DESTINATION_PREFIX; import static com.linkedin.datastream.common.DatastreamMetadataConstants.THROUGHPUT_VIOLATING_TOPICS; @@ -1425,7 +1426,25 @@ private void recordStreamProvisioningTime(Datastream ds) { } try { long streamProvisioningTimeMs = System.currentTimeMillis() - Long.parseLong(creationMsStr); - _log.info("Datastream {} transitioned to READY in {} ms", ds.getName(), streamProvisioningTimeMs); + + // Include the create-validation time (set by the resource provider via + // system.createValidationTime.ms) so the provisioning metric reflects validation time plus + // creation time. Absent or malformed is treated as 0, keeping the creation-time measurement + // intact for streams created before this property existed. + String validationMsStr = ds.getMetadata().get(CREATE_VALIDATION_TIME_MS); + if (validationMsStr != null) { + try { + long createValidationTimeMs = Long.parseLong(validationMsStr); + if (createValidationTimeMs > 0) { + streamProvisioningTimeMs += createValidationTimeMs; + } + } catch (NumberFormatException e) { + _log.warn("Invalid {} for datastream {}: {}", CREATE_VALIDATION_TIME_MS, ds.getName(), validationMsStr); + } + } + + _log.info("Datastream {} transitioned to READY in {} ms (creation + create-validation)", + ds.getName(), streamProvisioningTimeMs); if (streamProvisioningTimeMs >= 0) { _metrics.updateStreamProvisioningTimeHistogram(streamProvisioningTimeMs); // Emit the SLO counters atomically: the total provisioned and exactly one of within/outside SLA. diff --git a/datastream-server/src/main/java/com/linkedin/datastream/server/CoordinatorConfig.java b/datastream-server/src/main/java/com/linkedin/datastream/server/CoordinatorConfig.java index bda74277b..0bc01d1b2 100644 --- a/datastream-server/src/main/java/com/linkedin/datastream/server/CoordinatorConfig.java +++ b/datastream-server/src/main/java/com/linkedin/datastream/server/CoordinatorConfig.java @@ -59,7 +59,7 @@ public final class CoordinatorConfig { public static final int DEFAULT_MARK_DATASTREMS_STOPPED_TIMEOUT_MS = 60 * 1000; public static final int DEFAULT_MARK_DATASTREMS_STOPPED_RETRY_PERIOD_MS = 10 * 1000; public static final int DEFAULT_LOG_SIZE_LIMIT_IN_BYTES = 1024 * 1024; - public static final long DEFAULT_PROVISIONING_SLA_THRESHOLD_MS = Duration.ofMinutes(8).toMillis(); + public static final long DEFAULT_PROVISIONING_SLA_THRESHOLD_MS = Duration.ofMinutes(10).toMillis(); private final String _cluster; private final String _zkAddress; diff --git a/datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorConfig.java b/datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorConfig.java index 80ce3f0c3..77990aa36 100644 --- a/datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorConfig.java +++ b/datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorConfig.java @@ -5,6 +5,7 @@ */ package com.linkedin.datastream.server; +import java.time.Duration; import java.util.Properties; import org.testng.Assert; @@ -60,6 +61,19 @@ public void testTaskStopTimeoutAndRetryConfigDefault() { Assert.assertEquals(CoordinatorConfig.DEFAULT_TASK_STOP_CHECK_TIMEOUT_MS, config.getTaskStopCheckTimeoutMs()); } + @Test + public void testProvisioningSlaThresholdConfig() { + Properties props = new Properties(); + CoordinatorConfig config = createCoordinatorConfig(props); + Assert.assertEquals(config.getProvisioningSlaThresholdMs(), + CoordinatorConfig.DEFAULT_PROVISIONING_SLA_THRESHOLD_MS); + Assert.assertEquals(config.getProvisioningSlaThresholdMs(), Duration.ofMinutes(10).toMillis()); + + props.put(CoordinatorConfig.CONFIG_PROVISIONING_SLA_THRESHOLD_MS, "1000"); + CoordinatorConfig overridden = createCoordinatorConfig(props); + Assert.assertEquals(overridden.getProvisioningSlaThresholdMs(), 1000L); + } + @Test public void testForceStopStreamsOnFailureConfig() { Properties props = new Properties(); From 31fb0b41d76480b8000cc355d3ba1ca03701a55d Mon Sep 17 00:00:00 2001 From: Dhananjay Sawner Date: Mon, 22 Jun 2026 17:40:45 +0530 Subject: [PATCH 2/5] Make create-validation time addition explicit Compute createValidationTimeMs (defaulting to 0) and always add it, instead of conditionally adding inside the present-and-positive check. Logically identical, but the read-then-add-0-or-value flow is clearer. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../com/linkedin/datastream/server/Coordinator.java | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java index 6c952a4d8..bfe241194 100644 --- a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java +++ b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java @@ -1429,19 +1429,22 @@ private void recordStreamProvisioningTime(Datastream ds) { // Include the create-validation time (set by the resource provider via // system.createValidationTime.ms) so the provisioning metric reflects validation time plus - // creation time. Absent or malformed is treated as 0, keeping the creation-time measurement - // intact for streams created before this property existed. + // creation time. Read it from the metadata: add the value when present and greater than 0, + // otherwise add 0. This keeps the creation-time measurement intact for streams created before + // this property existed. + long createValidationTimeMs = 0L; String validationMsStr = ds.getMetadata().get(CREATE_VALIDATION_TIME_MS); if (validationMsStr != null) { try { - long createValidationTimeMs = Long.parseLong(validationMsStr); - if (createValidationTimeMs > 0) { - streamProvisioningTimeMs += createValidationTimeMs; + long parsedValidationTimeMs = Long.parseLong(validationMsStr); + if (parsedValidationTimeMs > 0) { + createValidationTimeMs = parsedValidationTimeMs; } } catch (NumberFormatException e) { _log.warn("Invalid {} for datastream {}: {}", CREATE_VALIDATION_TIME_MS, ds.getName(), validationMsStr); } } + streamProvisioningTimeMs += createValidationTimeMs; _log.info("Datastream {} transitioned to READY in {} ms (creation + create-validation)", ds.getName(), streamProvisioningTimeMs); From d37c915c8e2124943efe9a880898fb4e0b1beef6 Mon Sep 17 00:00:00 2001 From: Dhananjay Sawner Date: Mon, 22 Jun 2026 20:12:17 +0530 Subject: [PATCH 3/5] Refactoring --- .../datastream/common/DatastreamMetadataConstants.java | 4 +--- .../java/com/linkedin/datastream/server/Coordinator.java | 9 ++------- 2 files changed, 3 insertions(+), 10 deletions(-) diff --git a/datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamMetadataConstants.java b/datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamMetadataConstants.java index 87d10a81e..86babdc2a 100644 --- a/datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamMetadataConstants.java +++ b/datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamMetadataConstants.java @@ -90,9 +90,7 @@ public class DatastreamMetadataConstants { public static final String CREATION_MS = "system.creation.ms"; /** - * Time spent running create validation in the resource provider, in milliseconds. Added to the - * stream provisioning time alongside {@link #CREATION_MS}. Absent for streams created before this - * property existed, in which case it is treated as zero. + * Time spent running create validation in milliseconds. */ public static final String CREATE_VALIDATION_TIME_MS = "system.createValidationTime.ms"; diff --git a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java index bfe241194..2286343ac 100644 --- a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java +++ b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java @@ -1427,11 +1427,7 @@ private void recordStreamProvisioningTime(Datastream ds) { try { long streamProvisioningTimeMs = System.currentTimeMillis() - Long.parseLong(creationMsStr); - // Include the create-validation time (set by the resource provider via - // system.createValidationTime.ms) so the provisioning metric reflects validation time plus - // creation time. Read it from the metadata: add the value when present and greater than 0, - // otherwise add 0. This keeps the creation-time measurement intact for streams created before - // this property existed. + // Include the create-validation time in the stream provisioning time long createValidationTimeMs = 0L; String validationMsStr = ds.getMetadata().get(CREATE_VALIDATION_TIME_MS); if (validationMsStr != null) { @@ -1446,8 +1442,7 @@ private void recordStreamProvisioningTime(Datastream ds) { } streamProvisioningTimeMs += createValidationTimeMs; - _log.info("Datastream {} transitioned to READY in {} ms (creation + create-validation)", - ds.getName(), streamProvisioningTimeMs); + _log.info("Total stream provisioning time for Datastream {} - {} ms", ds.getName(), streamProvisioningTimeMs); if (streamProvisioningTimeMs >= 0) { _metrics.updateStreamProvisioningTimeHistogram(streamProvisioningTimeMs); // Emit the SLO counters atomically: the total provisioned and exactly one of within/outside SLA. From f152c5c191cc601140395bc360aaa94754a8cb23 Mon Sep 17 00:00:00 2001 From: Dhananjay Sawner Date: Mon, 22 Jun 2026 20:24:06 +0530 Subject: [PATCH 4/5] Extract create-validation time helper and unit-test it Move the metadata read into a @VisibleForTesting static helper getCreateValidationTimeMs(Datastream). Add TestCoordinatorStandalone, a general home for cluster-free Coordinator unit tests (the integration test stays as TestCoordinator in datastream-server-restli). It covers present-and-positive (included), absent, non-positive, and malformed (all excluded as 0). The inline logic lived in a private void method that writes to an internally-constructed metrics object, so it was not directly assertable; a pure static helper is. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../datastream/server/Coordinator.java | 33 ++++++----- .../server/TestCoordinatorStandalone.java | 58 +++++++++++++++++++ 2 files changed, 78 insertions(+), 13 deletions(-) create mode 100644 datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorStandalone.java diff --git a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java index 2286343ac..def31c734 100644 --- a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java +++ b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java @@ -1428,19 +1428,7 @@ private void recordStreamProvisioningTime(Datastream ds) { long streamProvisioningTimeMs = System.currentTimeMillis() - Long.parseLong(creationMsStr); // Include the create-validation time in the stream provisioning time - long createValidationTimeMs = 0L; - String validationMsStr = ds.getMetadata().get(CREATE_VALIDATION_TIME_MS); - if (validationMsStr != null) { - try { - long parsedValidationTimeMs = Long.parseLong(validationMsStr); - if (parsedValidationTimeMs > 0) { - createValidationTimeMs = parsedValidationTimeMs; - } - } catch (NumberFormatException e) { - _log.warn("Invalid {} for datastream {}: {}", CREATE_VALIDATION_TIME_MS, ds.getName(), validationMsStr); - } - } - streamProvisioningTimeMs += createValidationTimeMs; + streamProvisioningTimeMs += getCreateValidationTimeMs(ds); _log.info("Total stream provisioning time for Datastream {} - {} ms", ds.getName(), streamProvisioningTimeMs); if (streamProvisioningTimeMs >= 0) { @@ -1459,6 +1447,25 @@ private void recordStreamProvisioningTime(Datastream ds) { } } + /** + * Returns the create-validation time in milliseconds that is recorded in the + * datastream metadata under {@code system.createValidationTime.ms}, or 0 when the property is + * absent, not positive, or not a valid long. + */ + @VisibleForTesting + static long getCreateValidationTimeMs(Datastream ds) { + String validationMsStr = Objects.requireNonNull(ds.getMetadata()).get(CREATE_VALIDATION_TIME_MS); + if (validationMsStr == null) { + return 0L; + } + try { + long createValidationTimeMs = Long.parseLong(validationMsStr); + return createValidationTimeMs > 0 ? createValidationTimeMs : 0L; + } catch (NumberFormatException e) { + return 0L; + } + } + private void hardDeleteDatastream(Datastream ds, List activeStreams) { String taskPrefix; if (DatastreamUtils.containsTaskPrefix(ds)) { diff --git a/datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorStandalone.java b/datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorStandalone.java new file mode 100644 index 000000000..67830c128 --- /dev/null +++ b/datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorStandalone.java @@ -0,0 +1,58 @@ +/** + * Copyright 2019 LinkedIn Corporation. All rights reserved. + * Licensed under the BSD 2-Clause License. See the LICENSE file in the project root for license information. + * See the NOTICE file in the project root for additional information regarding copyright ownership. + */ +package com.linkedin.datastream.server; + +import org.testng.Assert; +import org.testng.annotations.Test; + +import com.linkedin.data.template.StringMap; +import com.linkedin.datastream.common.Datastream; + +import static com.linkedin.datastream.common.DatastreamMetadataConstants.CREATE_VALIDATION_TIME_MS; + + +/** + * Unit tests for {@link Coordinator} that run standalone, without an embedded cluster. The full + * integration-style test lives in {@code TestCoordinator} in the datastream-server-restli module. + */ +@Test +public class TestCoordinatorStandalone { + + private static Datastream datastreamWithValidationTime(String createValidationTimeMs) { + Datastream ds = new Datastream(); + StringMap metadata = new StringMap(); + if (createValidationTimeMs != null) { + metadata.put(CREATE_VALIDATION_TIME_MS, createValidationTimeMs); + } + ds.setMetadata(metadata); + return ds; + } + + @Test + public void testGetCreateValidationTimeMsPresent() { + // Present and positive -> included as-is + Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime("1500")), 1500L); + } + + @Test + public void testGetCreateValidationTimeMsAbsent() { + // Property not set -> excluded (0) + Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime(null)), 0L); + } + + @Test + public void testGetCreateValidationTimeMsNonPositive() { + // Zero or negative -> excluded (0) + Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime("0")), 0L); + Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime("-5")), 0L); + } + + @Test + public void testGetCreateValidationTimeMsMalformed() { + // Not a valid long -> excluded (0) + Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime("not-a-number")), 0L); + } +} From 32fe947a4861005843bbba8efa3e541dfa74766e Mon Sep 17 00:00:00 2001 From: Dhananjay Sawner Date: Tue, 23 Jun 2026 18:49:34 +0530 Subject: [PATCH 5/5] Log creation, create-validation, and total provisioning time Per review feedback, capture creation time and create-validation time separately and log all three when a stream reaches READY (creation + create-validation = total provisioning time). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../com/linkedin/datastream/server/Coordinator.java | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java index def31c734..b2a09b056 100644 --- a/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java +++ b/datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java @@ -1425,12 +1425,13 @@ private void recordStreamProvisioningTime(Datastream ds) { return; } try { - long streamProvisioningTimeMs = System.currentTimeMillis() - Long.parseLong(creationMsStr); + long creationTimeMs = System.currentTimeMillis() - Long.parseLong(creationMsStr); + long createValidationTimeMs = getCreateValidationTimeMs(ds); + // Total stream provisioning time is the creation time plus the create-validation time. + long streamProvisioningTimeMs = creationTimeMs + createValidationTimeMs; - // Include the create-validation time in the stream provisioning time - streamProvisioningTimeMs += getCreateValidationTimeMs(ds); - - _log.info("Total stream provisioning time for Datastream {} - {} ms", ds.getName(), streamProvisioningTimeMs); + _log.info("Stream provisioning time for Datastream {} - {} ms (creation {} ms + create-validation {} ms)", + ds.getName(), streamProvisioningTimeMs, creationTimeMs, createValidationTimeMs); if (streamProvisioningTimeMs >= 0) { _metrics.updateStreamProvisioningTimeHistogram(streamProvisioningTimeMs); // Emit the SLO counters atomically: the total provisioned and exactly one of within/outside SLA.