Add enhanced fan-out (EFO) support to ingestion-kinesis

[yuinumaz]

Description

Adds opt-in enhanced fan-out (EFO) support to the ingestion-kinesis plugin. With EFO, each primary shard reads records over a long-lived HTTP/2 SubscribeToShard push subscription with dedicated per-consumer throughput, instead of sharing the stream-wide GetRecords limits. The push subscription is bridged to the synchronous IngestionShardConsumer.readNext pull contract via a small demand-controlled queue (backpressure propagates to the subscription).

How to use

• enable_fanout: true (default false) opts in.
• Provide either:
  • fanout_consumer_arn — a pre-registered consumer ARN, used as-is, or
  • fanout_consumer_name — the plugin registers the consumer (or reuses it if it already exists) and caches the ARN per stream. Registration is idempotent, tolerant of concurrent registration, and waits for the consumer to become ACTIVE.
• If neither is set while fan-out is enabled, index creation fails with a clear error.

Notes

• Earliest/latest start positions map to the native SubscribeToShard TRIM_HORIZON / LATEST, so empty shards work correctly.
• Consumers are not deregistered automatically (matches KCL); operators clean up unused consumers.

Backward compatibility

• Default is off; the polling (GetRecords) path and all existing settings are unchanged. The existing fanout_consumer_arn behavior is preserved and takes precedence over auto-registration.

Testing

• Unit tests (mocked async/sync clients) cover the subscription bridge, resubscribe, start-position mapping, and consumer registration (reuse / register / race / not-active).
• ./gradlew :plugins:ingestion-kinesis:precommit :plugins:ingestion-kinesis:test passes.
• Verified end-to-end against a real Kinesis stream (4 shards): consumer auto-registered, records ingested and searchable.

Out of scope / follow-ups

• Automatic consumer deregistration.
• All-active ingestion sharing one consumer ARN.
• Resharding and Kinesis/OpenSearch shard-count mismatch handling.

Check List

• Functionality includes unit tests.
• Commits are signed per the DCO using --signoff.

[github-actions]

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit 19ed9c2.

⛔ Hard block: Issues at Medium severity or above will block this PR from merging.

Path	Line	Severity	Description
plugins/ingestion-kinesis/build.gradle	25	high	New dependency added: software.amazon.eventstream:eventstream:1.0.1. Per mandatory policy, all new dependencies must be flagged for maintainer verification regardless of apparent legitimacy.
plugins/ingestion-kinesis/build.gradle	75	high	New dependency added: org.reactivestreams:reactive-streams (pinned via existing versions.reactivestreams). Per mandatory policy, all new dependency additions must be flagged for maintainer verification.
plugins/ingestion-kinesis/src/main/plugin-metadata/plugin-security.policy	13	medium	Three new RuntimePermissions granted (accessDeclaredMembers, getClassLoader, setContextClassLoader) and write permission on aws.sharedCredentialsFile/aws.configFile system properties. These broaden the plugin's security sandbox and should be reviewed to confirm each permission is strictly required and cannot be abused by other code running in the same policy context.
plugins/ingestion-kinesis/src/main/java/org/opensearch/plugin/kinesis/KinesisShardConsumer.java	138	low	setDefaultAwsProfilePath() unconditionally redirects aws.sharedCredentialsFile and aws.configFile to the OpenSearch config directory via privileged System.setProperty calls. While the comment cites a legitimate reason (avoiding restricted ~/.aws access) and mirrors repository-s3 behavior, globally mutating JVM-wide system properties inside a privileged block warrants review to ensure no credential-loading side-effects on other components.

The table above displays the top 10 most important findings.

Total: 4 | Critical: 0 | High: 2 | Medium: 1 | Low: 1

---

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

---

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

[yuinumaz]

Thanks for the automated review. Addressing the Code-Diff-Analyzer findings:

New dependencies (high) — both are required for enhanced fan-out and cannot be dropped:

• software.amazon.eventstream:eventstream:1.0.1 — the AWS event-stream decoder the SDK uses to decode the HTTP/2 SubscribeToShard response stream. Without it, subscribing fails at runtime with NoClassDefFoundError: software.amazon.eventstream.MessageDecoder. It is part of the AWS SDK ecosystem; LICENSE/NOTICE/.jar.sha1 are included under plugins/ingestion-kinesis/licenses/.
• org.reactivestreams:reactive-streams — the Reactive Streams SPI implemented by the SDK's async SubscribeToShard publisher. Pinned via the existing versions.reactivestreams already used elsewhere in the repo.

Security policy (medium/low) — the pre-existing SocketPermission is unchanged. The new grants are the minimum needed for the AWS SDK v2 + Netty HTTP/2 client and follow the established repository-s3 pattern:

• accessDeclaredMembers / getClassLoader / setContextClassLoader for SDK/Netty reflection and class loading.
• PropertyPermission on aws.sharedCredentialsFile/aws.configFile + FilePermission "config","read" redirect the AWS profile path to the OpenSearch config dir so the SDK does not read the sandbox-restricted ~/.aws (identical approach to S3Service#setDefaultAwsProfilePath).
• I removed ReflectPermission "suppressAccessChecks" after verifying end-to-end (real Kinesis stream, records ingested and searchable) that it is not required.

Could a maintainer review and, if acceptable, apply the skip-diff-analyzer label? Happy to adjust.