aqibmehmoodDH opened a new pull request, #255: URL: https://github.com/apache/flink-connector-aws/pull/255
## Purpose of the change Related to https://issues.apache.org/jira/browse/FLINK-37908, which already has an open fix (#213) for the register/deregister race by making `deregisterStreamConsumer()` a permanent no-op. This PR takes a different approach that also closes a couple of related gaps found while investigating the same area, so I'm submitting it as an alternative for maintainers to weigh against #213 rather than assuming it should replace it. Three related problems in the `JOB_MANAGED` EFO consumer lifecycle: 1. **Register-time race**: `StreamConsumerRegistrar#registerStreamConsumer()` catches `ResourceInUseException` on conflict but never records `consumerArn` or verifies the consumer actually reached `ACTIVE`. A fast restart (any restart — task failure, redeploy, autoscale, not necessarily Kinesis-related) can land on a consumer that's still `DELETING` from the previous attempt's own deregistration, silently hand off an ARN that isn't actually usable, and leave `consumerArn` null so this attempt's own later deregistration becomes a no-op. 2. **No self-healing on missing consumer**: `KinesisStreamsSource#getConsumerArn()` retries `DescribeStreamConsumer` on `ResourceNotFoundException` but never attempts to recreate the consumer. Since nothing else in the connector re-registers on a task-level restart (the enumerator that owns registration isn't recreated by a pipelined-region restart), a consumer that goes missing for any reason — the race above, or external deletion — stays missing forever from the connector's perspective. 3. **Unbounded, zero-backoff retry, and a shard/consumer mixup**: `FanOutKinesisShardSubscription`'s recoverable-exception handling retried indefinitely with no backoff and no ceiling, so a permanently broken subscription (e.g. its consumer no longer exists) spun retrying forever with no visible task failure. Separately, its `ResourceNotFoundException` handling didn't distinguish "the shard is gone" (expected, e.g. resharding — should mark the split complete) from "the consumer is gone" (should fail the task so it can restart and self-heal) — both surface as the same exception type, and a raw `ResourceNotFoundException` is caught by `KinesisShardSplitReaderBase` and treated as the former in both cases, silently abandoning the shard when it's actually the latter. ## Brief change log - Added `StreamConsumerRegistrar#ensureActiveConsumer()`: describes the consumer; waits out an in-progress `DELETING` before creating a replacement; registers if missing; tolerates a concurrent `ResourceInUseException` on register as another caller/subtask winning the same race (converges by continuing to poll for `ACTIVE`, like everyone else); returns a **verified** `ACTIVE` ARN or throws after a bounded number of attempts. Used by both `registerStreamConsumer()` (so `consumerArn` is always correctly recorded) and `KinesisStreamsSource#getConsumerArn()` for `JOB_MANAGED` (so a missing consumer gets recreated instead of endlessly re-described). `SELF_MANAGED` behavior is unchanged — it still fails fast and never registers on the user's behalf. - Added bounded, full-jitter exponential backoff and a max-attempts ceiling to `FanOutKinesisShardSubscription`'s recoverable-exception retry, so a permanently broken subscription eventually throws instead of looping forever. - `ResourceNotFoundException` found anywhere in the cause chain (not just an exact top-level match — in practice it's frequently delivered wrapped, e.g. in `CompletionException`) is now checked against the exception message to distinguish "no such consumer" (wrapped in `KinesisStreamsSourceException`, fatal, triggers a task failure/restart) from "no such shard" (re-thrown raw as before, so `KinesisShardSplitReaderBase` still marks the split complete as it should for a genuinely closed shard). ## Verifying this change Added unit tests to `StreamConsumerRegistrarTest` (waits out `DELETING` before registering; tolerates a concurrent `ResourceInUseException` on register; `consumerArn` is recorded even on the conflict path) and `FanOutKinesisShardSubscriptionTest` (backoff is actually delayed, not immediate; eventually throws after the attempt ceiling; counter resets after a successful subscribe; a consumer-not-found `ResourceNotFoundException` is wrapped rather than rethrown raw, while a shard-not-found one still isn't). Also verified end-to-end against a live Kinesis stream (outside this repo's test suite): registering under real concurrent parallelism (10 parallel subtasks racing to register, one wins, the rest converge cleanly), waiting out a real in-progress deregistration before re-registering, and externally deregistering a running job's consumer and observing it fail fast at the next forced resubscription, restart, self-heal by registering a brand-new consumer, and resume reading — full recovery in ~16 seconds with no manual intervention. ## Significant changes - [ ] Dependencies have been added or upgraded - [ ] Public API has been changed (Public API is any class annotated with `@Public(Evolving)`) - [ ] Serializers have been changed - [ ] New feature has been introduced - If yes, how is this documented? (not applicable / docs / JavaDocs / not documented) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
