[vpj][dvc][cc][test] Create PubSub adapter factories based on VeniceProperties

[arjun4084346]

Problem Statement

In several places, Kafka admin/consumer/admin adapter factories are created directly (new ApacheKafkaAdminAdapterFactory()) without using venice properties. This obstructs using a different type of adapter transparently with the help of properties.
We are not passing configs needed by pubsub clients to VPJ.

Solution

• used PubSubClientsFactory.fromVeniceProperties to create adaptor factories
• passing pubsub configs obtained through pubsub broker wrappers to VPJ

Code changes

• Added new code behind a config. If so list the config names and their default values in the PR description.
• Introduced new log lines.
  • Confirmed if logs need to be rate limited to avoid excessive logging.

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).

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.

[Copilot]

Copilot reviewed 8 out of 8 changed files in this pull request and generated no comments.

Comments suppressed due to low confidence (2)

internal/venice-common/src/java/com/linkedin/venice/pubsub/PubSubClientsFactory.java:26

• [nitpick] Consider adding JavaDoc comments for the new generic parameters to improve code clarity and assist future maintainers.

public class PubSubClientsFactory<P extends PubSubProducerAdapter, C extends PubSubConsumerAdapter, A extends PubSubAdminAdapter> {

internal/venice-common/src/java/com/linkedin/venice/pubsub/PubSubClientsFactory.java:46

• [nitpick] Consider capturing explicit generic types in the factory method for better type inference and consistency across usages.

public static PubSubClientsFactory<? extends PubSubProducerAdapter, ? extends PubSubConsumerAdapter, ? extends PubSubAdminAdapter> fromVeniceProperties(

[sushantmane]

Even for draft PRs, having a clear and descriptive title is helpful.

[sushantmane]

Thanks, @arjun4084346! Let’s update the PR title to use “PubSub” instead of “Kafka” for clarity and accuracy:
"[vpj][dvc][cc][test] Create PubSub adapter factories based on VeniceProperties"

addressed comments

[sushantmane]

Overall, the changes look good. Just a few minor comments. Let’s also update the PR description and commit message to better reflect the current changes--they're a bit outdated now.

[sushantmane]

Thanks, @arjun4084346!