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..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 @@ -89,6 +89,11 @@ public class DatastreamMetadataConstants { */ public static final String CREATION_MS = "system.creation.ms"; + /** + * Time spent running create validation in milliseconds. + */ + 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..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 @@ -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; @@ -1424,8 +1425,13 @@ private void recordStreamProvisioningTime(Datastream ds) { return; } try { - long streamProvisioningTimeMs = System.currentTimeMillis() - Long.parseLong(creationMsStr); - _log.info("Datastream {} transitioned to READY in {} ms", ds.getName(), streamProvisioningTimeMs); + 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; + + _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. @@ -1442,6 +1448,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/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(); 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); + } +}