-
Notifications
You must be signed in to change notification settings - Fork 4.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
chore: load CDK clears partial aggregates at cadence (#48551)
- Loading branch information
Showing
16 changed files
with
852 additions
and
284 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
36 changes: 36 additions & 0 deletions
36
airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/state/TimeWindowTrigger.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| /* | ||
| * Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.cdk.load.state | ||
|
|
||
| import java.time.Clock | ||
|
|
||
| /* | ||
| * Simple time-windowing strategy for bucketing partial aggregates. | ||
| * | ||
| * Works off time relative to the injected @param clock. Generally this is the processing time domain. | ||
| */ | ||
| data class TimeWindowTrigger( | ||
| private val clock: Clock, | ||
| private val windowWidthMs: Long, | ||
| ) { | ||
| private var openedAtMs: Long? = null | ||
|
|
||
| /* | ||
| * Sets window open timestamp for computing completeness. Idempotent. Mutative. | ||
| */ | ||
| fun open(): Long { | ||
| if (openedAtMs == null) { | ||
| openedAtMs = clock.millis() | ||
| } | ||
| return openedAtMs!! | ||
| } | ||
|
|
||
| /* | ||
| * Returns whether window is complete relative to configured @param windowWidthMs. Non-mutative. | ||
| */ | ||
| fun isComplete(): Boolean { | ||
| return openedAtMs?.let { ts -> (clock.millis() - ts) >= windowWidthMs } ?: false | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
47 changes: 47 additions & 0 deletions
47
...yte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/task/internal/FlushTickTask.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| /* | ||
| * Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.cdk.load.task.internal | ||
|
|
||
| import com.google.common.annotations.VisibleForTesting | ||
| import io.airbyte.cdk.load.command.DestinationCatalog | ||
| import io.airbyte.cdk.load.command.DestinationStream | ||
| import io.airbyte.cdk.load.file.TimeProvider | ||
| import io.airbyte.cdk.load.message.DestinationStreamEvent | ||
| import io.airbyte.cdk.load.message.MessageQueueSupplier | ||
| import io.airbyte.cdk.load.message.StreamFlushEvent | ||
| import io.airbyte.cdk.load.state.Reserved | ||
| import io.airbyte.cdk.load.task.KillableScope | ||
| import io.airbyte.cdk.load.task.SyncLevel | ||
| import io.micronaut.context.annotation.Secondary | ||
| import io.micronaut.context.annotation.Value | ||
| import jakarta.inject.Singleton | ||
| import java.time.Clock | ||
|
|
||
| @Singleton | ||
| @Secondary | ||
| class FlushTickTask( | ||
| @Value("\${airbyte.flush.rate-ms}") private val tickIntervalMs: Long, | ||
| private val clock: Clock, | ||
| private val coroutineTimeUtils: TimeProvider, | ||
| private val catalog: DestinationCatalog, | ||
| private val recordQueueSupplier: | ||
| MessageQueueSupplier<DestinationStream.Descriptor, Reserved<DestinationStreamEvent>>, | ||
| ) : SyncLevel, KillableScope { | ||
| override suspend fun execute() { | ||
| while (true) { | ||
| waitAndPublishFlushTick() | ||
| } | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| suspend fun waitAndPublishFlushTick() { | ||
| coroutineTimeUtils.delay(tickIntervalMs) | ||
|
|
||
| catalog.streams.forEach { | ||
| val queue = recordQueueSupplier.get(it.descriptor) | ||
| queue.publish(Reserved(value = StreamFlushEvent(clock.millis()))) | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.