[server] Isolate consumers by store encryption policy

[minhmo1620]

Problem Statement

Server ingestion currently pools shared PubSub consumers only by resolved broker URL. When cluster.encryption.enabled is absent, that design cannot honor a target store encryption setting, and the first consumer created for a broker can incorrectly determine decryption behavior for other stores sharing the same broker.

The cluster property must remain authoritative when explicitly set to either true or false. Store-level fallback applies only when it is absent. Admin topics and metadata fetchers remain plaintext because they do not have a target store.

Solution

Resolve server-ingestion consumer decryption from the explicit cluster property when present; otherwise fall back to the target Store.isEncryptionEnabled() value. A missing store resolves to plaintext.

Shared consumer services are now keyed by a typed key containing the resolved broker URL and decryption mode. Creation returns the exact service selected for that policy, and subscription uses that service directly, preventing policy leakage between encrypted and plaintext stores on the same broker. Lookup, diagnostics, stale-consumer reporting, unsubscribe, and shutdown paths cover both pools.

This change depends on the configured PubSub consumer adapter honoring cluster.encryption.enabled: true must construct a decrypting/raw consumer path, while false must construct the normal plaintext consumer.

Code changes

• Added new code behind a config. Uses existing cluster.encryption.enabled; explicit true or false wins, while an absent property falls back to store encryption and defaults missing stores to plaintext.
• Introduced new log lines.
  • Confirmed if logs need to be rate limited to avoid excessive logging. The existing duplicate-service creation log now includes the typed pool key and remains on the bounded service-creation path.

Concurrency-Specific Checks

Both reviewer and PR author to verify

• Code has no race conditions or thread safety issues.
• Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed.
• No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
• Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList).
• Validated proper exception handling in multi-threaded code to avoid silent thread termination.

How was this PR tested?

• New unit tests added.
• New integration tests added.
• Modified or extended existing tests.
• Verified backward compatibility (if applicable).

Testing Done

• Local code review completed
• ./gradlew :clients:da-vinci-client:test --tests com.linkedin.davinci.kafka.consumer.AggKafkaConsumerServiceTest
• ./gradlew :clients:da-vinci-client:test --tests com.linkedin.davinci.kafka.consumer.StoreIngestionTaskTest --tests com.linkedin.davinci.kafka.consumer.KafkaStoreIngestionServiceTest
• ./gradlew spotlessJavaCheck :clients:da-vinci-client:assemble
• Commit hooks: spotlessApply and generateGHCI

Coverage includes explicit cluster true, explicit cluster false with an encrypted store, absent property with encrypted/plaintext/missing stores, mixed encrypted and plaintext stores sharing one broker, property propagation to consumer creation, reporting and cleanup across both modes, and creation failure propagation with a successful retry.

Does this PR introduce any user-facing or breaking changes?

• No. You can skip the rest of this section.
• Yes. Clearly explain the behavior change and its impact.

🤖 Generated with GitHub Copilot CLI

[Copilot]

🟡 Not ready to approve

There are a couple of concrete correctness/diagnostics issues (notably a potential NPE regression on Kafka URL resolution and diagnostic map-merging overwrites) that should be addressed before approval.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.

Pull request overview

This PR updates Da Vinci server ingestion to prevent shared PubSub consumer pools from leaking decryption/encryption policy across stores that share the same broker, by keying shared consumer services on both resolved broker URL and the resolved “decryption enabled” mode (derived from cluster.encryption.enabled when explicitly set, otherwise falling back to Store.isEncryptionEnabled()).

Changes:

• Resolve per-consumer decryption mode with explicit cluster property precedence and store-level fallback, defaulting missing stores to plaintext.
• Key AggKafkaConsumerService’s shared consumer-service pools by (resolvedKafkaUrl, decryptionEnabled) and route subscriptions/diagnostics across both pools.
• Add/extend unit tests to cover precedence, isolation across encrypted/plaintext stores on the same broker, and retryable creation failures.

File summaries

File	Description
clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/AggKafkaConsumerService.java	Introduces typed consumer-service key including decryption mode; updates lookup, subscribe, metrics, and diagnostics to handle both pools.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java	Resolves cluster.encryption.enabled with store fallback and stamps the resolved value into consumer properties; subscription uses the exact selected consumer service.
clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/AggKafkaConsumerServiceTest.java	Adds tests for config precedence, pool isolation, retryable creation failure, and diagnostics across both consumer modes.

Review details

Suppressed comments (1)

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/AggKafkaConsumerService.java:719

• In getIngestionInfoFor(..., regionName), merging results with putAll(...) can overwrite entries if both consumer modes return the same PubSubTopicPartition key, which would hide one mode's diagnostics. Consider merging with putIfAbsent(...) and logging when a duplicate key is encountered.

      if (consumerService != null) {
        consumerServiceFound = true;
        topicPartitionIngestionInfoMap
            .putAll(consumerService.getIngestionInfoFor(versionTopic, pubSubTopicPartition, true));
      }

• Files reviewed: 3/3 changed files
• Comments generated: 2
• Review effort level: Lite

We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.