diff --git a/config/config.eventbridge.extended.hocon b/config/config.eventbridge.extended.hocon index 0e74ff0c..782aaf98 100644 --- a/config/config.eventbridge.extended.hocon +++ b/config/config.eventbridge.extended.hocon @@ -65,6 +65,19 @@ # Optional, endpoint url configuration to override aws cloudwatch endpoint for metrics # Can be used to specify local endpoint when using localstack # "cloudwatchCustomEndpoint": "http://localhost:4566" + + # Optional, controls the KCL 2.x/3.x compatibility mode, used while migrating from + # enrich-eventbridge built with kinesis-client 2.x to this version (built with kinesis-client 3.x). + # Must be COMPATIBLE_WITH_2X_PHASE1, COMPATIBLE_WITH_2X or 3X. + # COMPATIBLE_WITH_2X_PHASE1: keeps the exact same behaviour and lease-balancing algorithm as + # kinesis-client 2.x. Deploy every worker with this setting first: the migration does not + # start yet, so it is safe to roll out gradually or roll back. + # COMPATIBLE_WITH_2X: once every worker runs COMPATIBLE_WITH_2X_PHASE1, switch to this value to + # start the actual migration to the KCL 3.x lease-balancing algorithm. + # 3X: full KCL 3.x mode. KCL switches to this automatically once the migration above has + # completed on every worker, so it should not normally need to be set explicitly. + # See https://docs.aws.amazon.com/streams/latest/dev/kcl-migration-from-2-3.html + "clientVersionConfig": "COMPATIBLE_WITH_2X_PHASE1" } "output": { diff --git a/config/config.kinesis.extended.hocon b/config/config.kinesis.extended.hocon index 238926a7..8c863ee3 100644 --- a/config/config.kinesis.extended.hocon +++ b/config/config.kinesis.extended.hocon @@ -65,6 +65,19 @@ # Optional, endpoint url configuration to override aws cloudwatch endpoint for metrics # Can be used to specify local endpoint when using localstack # "cloudwatchCustomEndpoint": "http://localhost:4566" + + # Optional, controls the KCL 2.x/3.x compatibility mode, used while migrating from + # enrich-kinesis built with kinesis-client 2.x to this version (built with kinesis-client 3.x). + # Must be COMPATIBLE_WITH_2X_PHASE1, COMPATIBLE_WITH_2X or 3X. + # COMPATIBLE_WITH_2X_PHASE1: keeps the exact same behaviour and lease-balancing algorithm as + # kinesis-client 2.x. Deploy every worker with this setting first: the migration does not + # start yet, so it is safe to roll out gradually or roll back. + # COMPATIBLE_WITH_2X: once every worker runs COMPATIBLE_WITH_2X_PHASE1, switch to this value to + # start the actual migration to the KCL 3.x lease-balancing algorithm. + # 3X: full KCL 3.x mode. KCL switches to this automatically once the migration above has + # completed on every worker, so it should not normally need to be set explicitly. + # See https://docs.aws.amazon.com/streams/latest/dev/kcl-migration-from-2-3.html + "clientVersionConfig": "COMPATIBLE_WITH_2X_PHASE1" } "output": { diff --git a/docs/KCL_3_MIGRATION.MD b/docs/KCL_3_MIGRATION.MD new file mode 100644 index 00000000..b3cf08cb --- /dev/null +++ b/docs/KCL_3_MIGRATION.MD @@ -0,0 +1,253 @@ +# Migrating enrich-kinesis / enrich-eventbridge from KCL 2.x to KCL 3.5 + +This document is for maintainers operating `enrich-kinesis` / `enrich-eventbridge` deployments that +are moving from `amazon-kinesis-client` 2.7.3 to 3.5.1 (this repo skips the 3.0–3.4 line and jumps +straight to 3.5, which uses the **single table format** by default for new 2.x → 3.5 migrations). + +It covers: +1. What changed in the codebase and config +2. The rollout procedure (application-side) +3. AWS-side prerequisites (IAM, EC2/ECS/EKS metadata requirements) +4. CloudWatch metrics/alarms worth watching during and after migration +5. A note on the `MetricsLevel` vs metric-dimensions distinction (and why this migration doesn't + change either) + +Reference: [AWS docs - Migrate from KCL 2.x to KCL 3.x](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration-from-2-3.html), +[Single table format for KCL](https://docs.aws.amazon.com/streams/latest/dev/kcl-single-table-format.html), +[IAM permissions for KCL](https://docs.aws.amazon.com/streams/latest/dev/kcl-iam-permissions.html), +[Monitor KCL with CloudWatch](https://docs.aws.amazon.com/streams/latest/dev/monitoring-with-kcl.html). + +--- + +## 1. What changed in this codebase + +- `amazon-kinesis-client` bumped to `3.5.1` (`project/Dependencies.scala`). +- A new input config field, `clientVersionConfig`, was added to `Input.Kinesis` + (`modules/common-fs2/.../config/io.scala`). It maps directly onto + `CoordinatorConfig.ClientVersionConfig` and controls the 2.x/3.x compatibility mode: + - `COMPATIBLE_WITH_2X_PHASE1` (default in `application.conf`) — behaves identically to KCL 2.x; + migration does **not** start yet. Safe to deploy/rollback freely. + - `COMPATIBLE_WITH_2X` — starts the actual migration to the KCL 3.x lease-balancing algorithm. + - `3X` — full KCL 3.x mode. KCL switches to this automatically once migration completes; you + should not normally need to set it explicitly. +- `Source.scala` (in `kinesis` and `eventbridge` modules) wires this into + `configsBuilder.coordinatorConfig.clientVersionConfig(...)`. +- `MetricsConfig.metricsLevel` behaviour is unchanged from before this migration: `DETAILED` when + `monitoring.metrics.cloudwatch = true`, `NONE` otherwise. See [section 5](#5-a-note-on-metricslevel-vs-metricsenableddimensions) + for a discussion of a *separate*, optional CloudWatch-dimension change that is **not** part of + this migration. +- Because this is a **direct** 2.x → 3.5 migration (not 2.x → 3.0-3.4 → 3.5), single table format is + used from day one. `ConfigsBuilder` still uses `appName` as the lease table name, so your existing + lease table is reused as-is; KCL layers worker-metrics/coordinator-state entries into that same + table (distinguished by an `entityType` attribute) instead of creating separate tables. **No lease + table schema migration is required and no separate lease-stealing/duplicate-processing risk is + introduced by the table structure itself** — the risk is purely in the coordination protocol + during the phase1 → phase2 switch below. + +## 2. Rollout procedure + +> **Critical**: all workers of a given KCL application must run the *same* `clientVersionConfig` +> mode at (almost) the same time. KCL 2.x and KCL 3.x use different lease-balancing algorithms; +> running a mix for an extended period causes them to fight over lease ownership, which is the +> actual source of duplicate/re-processed records during migration — not the DynamoDB table +> structure. + +### Phase 0 — Deploy KCL 3.5 code, `COMPATIBLE_WITH_2X_PHASE1` (already the default) + +- Deploy the new build (this repo already defaults `clientVersionConfig` to + `COMPATIBLE_WITH_2X_PHASE1` in `application.conf`). +- Behaviour is functionally identical to KCL 2.7.3. This is safe to roll out gradually + (canary/rolling deploy) and safe to roll back at any time. +- Confirm IAM permissions are updated first (see [section 3](#3-aws-side-prerequisites)) — the new + code will fail on lease-table `UpdateTable`/GSI `Query` calls otherwise. + +### Phase 1 — Bake and confirm + +- Let all workers run on `COMPATIBLE_WITH_2X_PHASE1` for a while (at least one full deploy cycle + across your whole fleet). +- Watch the `Migration` operation metrics in CloudWatch (namespace = your KCL `appName`): + - `CurrentState:2xCompatibleWorker` (Sum) should equal your total worker count. + - `Fault` should be at/near zero. +- Do **not** proceed to Phase 2 until every worker is confirmed running the new code. + +### Phase 2 — Flip to `COMPATIBLE_WITH_2X` to start the migration + +- Change `input.clientVersionConfig` to `COMPATIBLE_WITH_2X` and roll out to **all** workers. +- This is what actually triggers the switch to the KCL 3.x resource-utilization-based + lease-balancing algorithm. +- A rolling deploy is fine, but complete it promptly — don't leave the fleet in a mixed state for + a long time. + +### Phase 3 — Watch migration complete + +- Keep watching the `Migration` operation metrics: + - `CurrentState:3xWorker` (Sum) rises as workers switch over. + - When `CurrentState:3xWorker` equals your total worker count, migration is functionally + complete on the coordination side. + - Allow **at least 10 minutes** after that point before considering it fully settled — KCL needs + this time to finish switching every worker to the new algorithm. + - `GsiStatusReady` should show the lease table GSI is available (required for the new + `LeaseDiscovery` operation). + +### Phase 4 — Leave it at `COMPATIBLE_WITH_2X` (or optionally clean up to `3X`) + +- You do not need to manually set `3X`; KCL detects completion and behaves like KCL 3.x internally + regardless. Leaving the config at `COMPATIBLE_WITH_2X` is fine indefinitely. +- Optionally, once you're confident the fleet has fully migrated and you don't need the ability to + roll back to 2.x-compatible behaviour, update the default in `application.conf` / + `config/config.kinesis.extended.hocon` / `config/config.eventbridge.extended.hocon` from + `COMPATIBLE_WITH_2X_PHASE1` to `COMPATIBLE_WITH_2X` so fresh environments skip phase 1. + +### Rollback considerations + +- **Phase 0/1** (`COMPATIBLE_WITH_2X_PHASE1`): safe to roll back to KCL 2.7.3 code at any time. +- **Phase 2** (`COMPATIBLE_WITH_2X`, migration in progress): you can revert the config back to + `COMPATIBLE_WITH_2X_PHASE1` and workers will fall back to KCL 2.x-style balancing. +- **After full migration completes**: rollback to 2.x-compatible behaviour is no longer meaningful. + +--- + +## 3. AWS-side prerequisites + +### 3.1 IAM permissions + +KCL 3.x needs a few more IAM actions than 2.x did. Update the IAM role/policy attached to your +enrich workers **before** deploying the new code. Replace `REGION`, `ACCOUNT_ID`, `STREAM_NAME` and +`KCL_APPLICATION_NAME` (this is `input.appName` in enrich's config, i.e. the DynamoDB lease table +name) with your actual values: + +```json +{ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": [ + "kinesis:DescribeStream", + "kinesis:DescribeStreamSummary", + "kinesis:RegisterStreamConsumer", + "kinesis:GetRecords", + "kinesis:GetShardIterator", + "kinesis:ListShards" + ], + "Resource": "arn:aws:kinesis:REGION:ACCOUNT_ID:stream/STREAM_NAME" + }, + { + "Effect": "Allow", + "Action": [ + "kinesis:SubscribeToShard", + "kinesis:DescribeStreamConsumer" + ], + "Resource": "arn:aws:kinesis:REGION:ACCOUNT_ID:stream/STREAM_NAME/consumer/*" + }, + { + "Effect": "Allow", + "Action": [ + "dynamodb:CreateTable", + "dynamodb:DescribeTable", + "dynamodb:UpdateTable", + "dynamodb:GetItem", + "dynamodb:UpdateItem", + "dynamodb:PutItem", + "dynamodb:DeleteItem", + "dynamodb:Scan" + ], + "Resource": [ + "arn:aws:dynamodb:REGION:ACCOUNT_ID:table/KCL_APPLICATION_NAME" + ] + }, + { + "Effect": "Allow", + "Action": [ + "dynamodb:Query" + ], + "Resource": [ + "arn:aws:dynamodb:REGION:ACCOUNT_ID:table/KCL_APPLICATION_NAME/index/*" + ] + }, + { + "Effect": "Allow", + "Action": [ + "cloudwatch:PutMetricData" + ], + "Resource": "*" + } + ] +} +``` + +Notes: +- The `dynamodb:UpdateTable` and GSI `dynamodb:Query` actions are **new requirements versus KCL + 2.x** — 2.x never needed to alter the lease table schema or query a GSI on it. Missing these is + the most common cause of KCL 3.x workers failing to start. +- Because this repo migrates straight to **single table format**, you do **not** need the separate + `KCL_APPLICATION_NAME-WorkerMetricStats` / `KCL_APPLICATION_NAME-CoordinatorState` table + permissions that AWS's generic KCL 3.x docs mention for 3.0–3.4 style migrations — those tables + are never created in a direct 2.x → 3.5 migration. If you ever do end up with those tables (e.g. + a past 3.0–3.4 deployment before this migration), also grant `CreateTable`, `DescribeTable`, + `Scan`, `GetItem`, `PutItem`, `UpdateItem`, `DeleteItem` and (for cleanup) `DeleteTable` on them. +- `cloudwatch:PutMetricData` is required regardless of `metricsLevel`/`monitoring.metrics.cloudwatch` + settings if you ever enable CloudWatch metrics reporting. + +### 3.2 Worker CPU-utilization metrics prerequisites + +KCL 3.x's lease-balancing algorithm uses worker CPU utilization to balance load. If it can't +collect this, it silently falls back to balancing purely on lease/throughput count per worker +(still correct, just less optimal). Requirements per platform: + +- **EC2**: Linux OS, [IMDSv2](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html) enabled. +- **ECS on EC2**: Linux OS, [ECS task metadata endpoint v4](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ec2-metadata.html) enabled, ECS agent >= 1.39.0. +- **ECS on Fargate**: [Fargate task metadata endpoint v4](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-metadata-endpoint-v4-fargate.html) (default on platform >= 1.4.0). +- **EKS on EC2**: Linux OS. +- **EKS on Fargate**: platform >= 1.3.0. + +Check your Terraform/deployment tooling enforces IMDSv2 and the relevant metadata endpoint version +before migrating. + +--- + +## 4. Helpful CloudWatch metrics/alarms + +### 4.1 Migration-specific (temporary, remove/ignore once migration is complete) + +All under the `Migration` operation, `SUMMARY` level, namespace = your `appName`: + +| Metric | What to alarm on | +| --- | --- | +| `CurrentState:2xCompatibleWorker` | Track during Phase 1 bake — should equal total worker count before proceeding to Phase 2. | +| `CurrentState:3xWorker` | Track during/after Phase 2 — should reach total worker count; this is your "migration complete" signal. | +| `Fault` | Alarm if `Sum > 0` sustained for several periods — indicates the migration state machine is hitting errors (often transient/self-retried, but a persistent non-zero value warrants investigation). | +| `GsiStatusReady` | Confirms the lease table's GSI is available; should be `1` throughout. | + +### 4.2 Ongoing operational health (useful beyond migration) + +| Metric (scope) | Suggested alarm | +| --- | --- | +| `ProcessTask.MillisBehindLatest` (per-shard/aggregate, `SUMMARY`) | Alarm on sustained high values — indicates the consumer is falling behind the stream. This is the single most important consumer-lag signal. | +| `LeaseAssignmentManager.TotalLeases` vs actual shard count | Sanity check — mismatch indicates leases aren't being created/synced correctly. | +| `LeaseAssignmentManager.LeaseSpillover` | Non-zero sustained values mean you don't have enough workers/capacity for the number of leases — consider scaling out. | +| `RenewAllLeases.LostLeases` | Sustained non-zero indicates workers are losing leases unexpectedly (crashes, GC pauses, network issues) — a leading indicator of potential reprocessing. | +| `LeaseAssignmentManager.NumWorkers` | Cross-check against your actual deployed worker count — mismatches can indicate zombie/leaked workers still holding leases. | +| DynamoDB `ThrottledRequests` / `ConsumedWriteCapacityUnits` on the lease table (native DynamoDB metric, not KCL) | Since single table format concentrates lease + worker-metrics + coordinator-state writes into one table, watch DynamoDB throttling more closely than before — consider on-demand billing mode or adequate provisioned capacity headroom during migration. | + +Since `WorkerIdentifier` restarts with a new value unless you pin worker IDs, avoid alarming +directly on per-worker series unless your deployment assigns stable worker IDs across restarts. + +--- + +## 5. Quick checklist + +- [ ] AWS SDK for Java pinned to >= 2.28.0 (not 2.27.19–2.27.23). +- [ ] IAM policy updated with `UpdateTable`, GSI `Query`, and (if applicable) legacy + WorkerMetricStats/CoordinatorState table permissions. +- [ ] IMDSv2 / ECS-EKS metadata endpoint v4 enabled where applicable. +- [ ] Deploy with `clientVersionConfig = COMPATIBLE_WITH_2X_PHASE1` (default) to 100% of fleet. +- [ ] Confirm `CurrentState:2xCompatibleWorker` == total worker count in CloudWatch. +- [ ] Flip to `clientVersionConfig = COMPATIBLE_WITH_2X`, deploy to 100% of fleet promptly. +- [ ] Confirm `CurrentState:3xWorker` == total worker count, wait >= 10 minutes. +- [ ] Confirm no unexpected DynamoDB throttling occurred on the lease table during migration. +- [ ] (Optional) Flip default config to `COMPATIBLE_WITH_2X` for future fresh deployments. +- [ ] (Team decision, optional, separate patch) Discuss and apply the standalone + `shardLevelMetricsEnabled` patch if per-shard CloudWatch metric/alarm sprawl is a real + problem; separately decide on `metricsLevel` (`DETAILED` vs `SUMMARY`) as a cost/verbosity + trade-off. diff --git a/modules/common-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/io.scala b/modules/common-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/io.scala index 19072044..d0745818 100644 --- a/modules/common-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/io.scala +++ b/modules/common-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/io.scala @@ -130,7 +130,8 @@ object io { checkpointBackoff: BackoffPolicy, customEndpoint: Option[URI], dynamodbCustomEndpoint: Option[URI], - cloudwatchCustomEndpoint: Option[URI] + cloudwatchCustomEndpoint: Option[URI], + clientVersionConfig: Kinesis.ClientVersionConfig ) extends Input with RetryCheckpointing @@ -194,6 +195,38 @@ object io { implicit val retrievalEncoder: Encoder[Retrieval] = deriveConfiguredEncoder[Retrieval] } + // Controls the KCL 2.x/3.x compatibility mode used while migrating the lease/checkpoint table. + // See https://docs.aws.amazon.com/streams/latest/dev/kcl-migration-from-2-3.html + sealed trait ClientVersionConfig + object ClientVersionConfig { + // Deploy every worker with this setting first. The application keeps behaving exactly like + // KCL 2.x (same lease-balancing algorithm) and the migration to KCL 3.x does not start yet. + case object CompatibleWith2xPhase1 extends ClientVersionConfig + // Once every worker is running with CompatibleWith2xPhase1, switch to this setting to kick off + // the actual migration to the KCL 3.x lease-balancing algorithm. + case object CompatibleWith2x extends ClientVersionConfig + // Full KCL 3.x mode. KCL automatically switches to this once the migration above has completed, + // so it should not normally need to be set explicitly. + case object Kcl3x extends ClientVersionConfig + + implicit val clientVersionConfigDecoder: Decoder[ClientVersionConfig] = + Decoder.decodeString.emap { raw => + raw.toUpperCase match { + case "COMPATIBLE_WITH_2X_PHASE1" => CompatibleWith2xPhase1.asRight + case "COMPATIBLE_WITH_2X" => CompatibleWith2x.asRight + case "3X" => Kcl3x.asRight + case _ => + s"clientVersionConfig $raw is not supported. Possible types are COMPATIBLE_WITH_2X_PHASE1, COMPATIBLE_WITH_2X and 3X".asLeft + } + } + implicit val clientVersionConfigEncoder: Encoder[ClientVersionConfig] = + Encoder.encodeString.contramap { + case CompatibleWith2xPhase1 => "COMPATIBLE_WITH_2X_PHASE1" + case CompatibleWith2x => "COMPATIBLE_WITH_2X" + case Kcl3x => "3X" + } + } + implicit val kinesisDecoder: Decoder[Kinesis] = deriveConfiguredDecoder[Kinesis] implicit val kinesisEncoder: Encoder[Kinesis] = deriveConfiguredEncoder[Kinesis] } diff --git a/modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/IntegrationTestConfig.scala b/modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/IntegrationTestConfig.scala index d8da1b2d..bfbe0b7c 100644 --- a/modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/IntegrationTestConfig.scala +++ b/modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/IntegrationTestConfig.scala @@ -59,7 +59,8 @@ object IntegrationTestConfig { BackoffPolicy(10.millis, 10.seconds, Some(10)), Some(URI.create(getEndpoint(localstackPort))), Some(URI.create(getEndpoint(localstackPort))), - Some(URI.create(getEndpoint(localstackPort))) + Some(URI.create(getEndpoint(localstackPort))), + Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 ) val monitoring = Monitoring( diff --git a/modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/Source.scala b/modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/Source.scala index d39988f1..d78be829 100644 --- a/modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/Source.scala +++ b/modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/Source.scala @@ -32,7 +32,7 @@ import eu.timepit.refined.numeric._ import software.amazon.awssdk.regions.Region import software.amazon.kinesis.common.{ConfigsBuilder, InitialPositionInStream, InitialPositionInStreamExtended} import software.amazon.kinesis.processor.SingleStreamTracker -import software.amazon.kinesis.coordinator.Scheduler +import software.amazon.kinesis.coordinator.{CoordinatorConfig, Scheduler} import software.amazon.kinesis.metrics.MetricsLevel import software.amazon.kinesis.processor.ShardRecordProcessorFactory import software.amazon.kinesis.retrieval.polling.PollingConfig @@ -144,9 +144,21 @@ object Source { if (monitoring.metrics.cloudwatch) MetricsLevel.DETAILED else MetricsLevel.NONE } + val coordinatorConfig = configsBuilder.coordinatorConfig + .clientVersionConfig { + kinesisConfig.clientVersionConfig match { + case Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 => + CoordinatorConfig.ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X_PHASE1 + case Input.Kinesis.ClientVersionConfig.CompatibleWith2x => + CoordinatorConfig.ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X + case Input.Kinesis.ClientVersionConfig.Kcl3x => + CoordinatorConfig.ClientVersionConfig.CLIENT_VERSION_CONFIG_3X + } + } + new Scheduler( configsBuilder.checkpointConfig, - configsBuilder.coordinatorConfig, + coordinatorConfig, configsBuilder.leaseManagementConfig, configsBuilder.lifecycleConfig, metricsConfig, diff --git a/modules/eventbridge/src/main/resources/application.conf b/modules/eventbridge/src/main/resources/application.conf index e1472342..0f603690 100644 --- a/modules/eventbridge/src/main/resources/application.conf +++ b/modules/eventbridge/src/main/resources/application.conf @@ -15,6 +15,7 @@ "maxBackoff": 10 seconds "maxRetries": 10 } + "clientVersionConfig": "COMPATIBLE_WITH_2X_PHASE1" } "output": { diff --git a/modules/eventbridge/src/main/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/Source.scala b/modules/eventbridge/src/main/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/Source.scala index dbeabd99..0a6601f0 100644 --- a/modules/eventbridge/src/main/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/Source.scala +++ b/modules/eventbridge/src/main/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/Source.scala @@ -28,7 +28,7 @@ import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient import software.amazon.awssdk.services.kinesis.KinesisAsyncClient import software.amazon.kinesis.common.{ConfigsBuilder, InitialPositionInStream, InitialPositionInStreamExtended} import software.amazon.kinesis.processor.SingleStreamTracker -import software.amazon.kinesis.coordinator.Scheduler +import software.amazon.kinesis.coordinator.{CoordinatorConfig, Scheduler} import software.amazon.kinesis.metrics.MetricsLevel import software.amazon.kinesis.processor.ShardRecordProcessorFactory import software.amazon.kinesis.retrieval.fanout.FanOutConfig @@ -137,9 +137,21 @@ object Source { if (monitoring.metrics.cloudwatch) MetricsLevel.DETAILED else MetricsLevel.NONE } + val coordinatorConfig = configsBuilder.coordinatorConfig + .clientVersionConfig { + kinesisConfig.clientVersionConfig match { + case Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 => + CoordinatorConfig.ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X_PHASE1 + case Input.Kinesis.ClientVersionConfig.CompatibleWith2x => + CoordinatorConfig.ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X + case Input.Kinesis.ClientVersionConfig.Kcl3x => + CoordinatorConfig.ClientVersionConfig.CLIENT_VERSION_CONFIG_3X + } + } + new Scheduler( configsBuilder.checkpointConfig, - configsBuilder.coordinatorConfig, + coordinatorConfig, configsBuilder.leaseManagementConfig, configsBuilder.lifecycleConfig, metricsConfig, diff --git a/modules/eventbridge/src/test/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/ConfigSpec.scala b/modules/eventbridge/src/test/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/ConfigSpec.scala index 990ff220..c2ed0f80 100644 --- a/modules/eventbridge/src/test/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/ConfigSpec.scala +++ b/modules/eventbridge/src/test/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/ConfigSpec.scala @@ -47,7 +47,8 @@ class ConfigSpec extends Specification with CatsEffect { io.BackoffPolicy(100.milli, 10.second, Some(10)), None, None, - None + None, + io.Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 ), io.Outputs( io.Output.Eventbridge( @@ -130,7 +131,8 @@ class ConfigSpec extends Specification with CatsEffect { io.BackoffPolicy(100.milli, 10.second, Some(10)), None, None, - None + None, + io.Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 ), io.Outputs( io.Output.Eventbridge( diff --git a/modules/kinesis/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/KinesisConfig.scala b/modules/kinesis/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/KinesisConfig.scala index d5dee282..0c27a7fb 100644 --- a/modules/kinesis/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/KinesisConfig.scala +++ b/modules/kinesis/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/KinesisConfig.scala @@ -35,7 +35,8 @@ object KinesisConfig { BackoffPolicy(10.millis, 10.seconds, Some(10)), Some(URI.create(getEndpoint(localstackPort))), Some(URI.create(getEndpoint(localstackPort))), - Some(URI.create(getEndpoint(localstackPort))) + Some(URI.create(getEndpoint(localstackPort))), + Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 ) def badStreamConfig(localstackPort: Int, streamName: String) = @@ -49,7 +50,8 @@ object KinesisConfig { BackoffPolicy(10.millis, 10.seconds, Some(10)), Some(URI.create(getEndpoint(localstackPort))), Some(URI.create(getEndpoint(localstackPort))), - Some(URI.create(getEndpoint(localstackPort))) + Some(URI.create(getEndpoint(localstackPort))), + Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 ) def rawStreamConfig(localstackPort: Int, streamName: String) = diff --git a/modules/kinesis/src/main/resources/application.conf b/modules/kinesis/src/main/resources/application.conf index f93d125d..f72bd70a 100644 --- a/modules/kinesis/src/main/resources/application.conf +++ b/modules/kinesis/src/main/resources/application.conf @@ -15,6 +15,7 @@ "maxBackoff": 10 seconds "maxRetries": 10 } + "clientVersionConfig": "COMPATIBLE_WITH_2X_PHASE1" } "output": { diff --git a/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/enrich/kinesis/Source.scala b/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/enrich/kinesis/Source.scala index e1db0111..2da24599 100644 --- a/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/enrich/kinesis/Source.scala +++ b/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/enrich/kinesis/Source.scala @@ -32,7 +32,7 @@ import eu.timepit.refined.numeric._ import software.amazon.awssdk.regions.Region import software.amazon.kinesis.common.{ConfigsBuilder, InitialPositionInStream, InitialPositionInStreamExtended} import software.amazon.kinesis.processor.SingleStreamTracker -import software.amazon.kinesis.coordinator.Scheduler +import software.amazon.kinesis.coordinator.{CoordinatorConfig, Scheduler} import software.amazon.kinesis.metrics.MetricsLevel import software.amazon.kinesis.processor.ShardRecordProcessorFactory import software.amazon.kinesis.retrieval.polling.PollingConfig @@ -144,9 +144,21 @@ object Source { if (monitoring.metrics.cloudwatch) MetricsLevel.DETAILED else MetricsLevel.NONE } + val coordinatorConfig = configsBuilder.coordinatorConfig + .clientVersionConfig { + kinesisConfig.clientVersionConfig match { + case Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 => + CoordinatorConfig.ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X_PHASE1 + case Input.Kinesis.ClientVersionConfig.CompatibleWith2x => + CoordinatorConfig.ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X + case Input.Kinesis.ClientVersionConfig.Kcl3x => + CoordinatorConfig.ClientVersionConfig.CLIENT_VERSION_CONFIG_3X + } + } + new Scheduler( configsBuilder.checkpointConfig, - configsBuilder.coordinatorConfig, + coordinatorConfig, configsBuilder.leaseManagementConfig, configsBuilder.lifecycleConfig, metricsConfig, diff --git a/modules/kinesis/src/test/scala/com/snowplowanalytics/snowplow/enrich/kinesis/ConfigSpec.scala b/modules/kinesis/src/test/scala/com/snowplowanalytics/snowplow/enrich/kinesis/ConfigSpec.scala index df9c4c09..0aac66b9 100644 --- a/modules/kinesis/src/test/scala/com/snowplowanalytics/snowplow/enrich/kinesis/ConfigSpec.scala +++ b/modules/kinesis/src/test/scala/com/snowplowanalytics/snowplow/enrich/kinesis/ConfigSpec.scala @@ -47,7 +47,8 @@ class ConfigSpec extends Specification with CatsEffect { io.BackoffPolicy(100.milli, 10.second, Some(10)), None, None, - None + None, + io.Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 ), io.Outputs( io.Output.Kinesis( @@ -152,7 +153,8 @@ class ConfigSpec extends Specification with CatsEffect { io.BackoffPolicy(100.milli, 10.second, Some(10)), None, None, - None + None, + io.Input.Kinesis.ClientVersionConfig.CompatibleWith2xPhase1 ), io.Outputs( io.Output.Kinesis( diff --git a/project/Dependencies.scala b/project/Dependencies.scala index 00cc800f..c9ec8962 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -73,8 +73,8 @@ object Dependencies { val collectorPayload = "0.0.0" val schemaSniffer = "0.0.0" val gcpSdk = "2.62.1" - val awsSdk = "2.46.11" - val kinesisClient = "2.7.3" + val awsSdk = "2.47.5" + val kinesisClient = "3.5.1" val kafka = "3.9.2" val lz4Java = "1.10.4" val mskAuth = "2.3.7"