[ANA-4718] Namespace pattern subscriber#1
Merged
vbabenkoru merged 4 commits intov4.2-iterablefrom Mar 26, 2026
Merged
Conversation
…overy Introduces NamespacePatternSubscriber which discovers both namespaces and topics dynamically using regex patterns. This enables consuming from patterns like tenant/namespace-pattern/topic-pattern where both namespace and topic names are discovered at runtime. Key features: - Pattern format: tenant/namespace-regex/topic-regex - Auto-derives admin URL from service URL (pulsar://host:6650 -> http://host:8080) - Uses existing partition discovery interval for both namespace and topic discovery - Falls back to last known namespaces on discovery failures - Minimal PulsarAdmin support restored for namespace listing only This is essential for deployments where namespaces are created dynamically and need to be discovered automatically without manual configuration updates.
Moved PulsarAdmin creation from NamespacePatternSubscriber to PulsarSourceEnumerator to ensure proper resource cleanup and follow dependency injection pattern. Previously, PulsarAdmin instances were never closed, causing connection pool exhaustion in long-running applications. Key changes: - Add RequiresPulsarAdmin interface for capability-based injection - Enumerator now owns and manages both PulsarClient and PulsarAdmin lifecycles - Fix resource leak: PulsarAdmin now properly closed in enumerator.close() - Remove code duplication: deriveAdminUrl() consolidated in PulsarClientFactory - Add missing TLS keystore options to PulsarClientFactory.createAdmin() - Fix thread safety: use CopyOnWriteArrayList for namespace caching - Fix bounds checking to prevent IndexOutOfBoundsException in topic/namespace parsing
The lastKnownNamespaces field is transient and becomes null after deserialization. The error handling code was calling .isEmpty() without checking for null first, causing a NullPointerException that masked the original exception. This made debugging impossible when namespace discovery failed after a restart or failover. Added null check before isEmpty() to properly handle the deserialized state.
mallize
approved these changes
Mar 26, 2026
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
mallize
approved these changes
Mar 26, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose of the change
Add a
NamespacePatternSubscriberthat dynamically discovers namespaces and topics using a pattern in the formattenant/namespace-pattern/topic-pattern. This enables consuming from topics across multiple namespaces without explicitly listing each one — useful when namespaces are created dynamically (e.g., per-organization).Brief change log
NamespacePatternSubscriberthat discovers namespaces matching a regex under a tenant via PulsarAdmin, then discovers topics matching a second regex within each namespace using Pulsar's lookup protocol.RequiresPulsarAdmincapability interface so the enumerator can inject aPulsarAdmininstance into subscribers that need administrative operations (e.g., listing namespaces).PulsarClientFactory.createAdmin()to create aPulsarAdminfrom Flink configuration, with automatic URL derivation from the service URL whenPULSAR_ADMIN_URLis not set.PULSAR_ADMIN_URLconfig option toPulsarOptions.setNamespaceTopicPattern()builder methods toPulsarSourceBuilder.PulsarAdminlifecycle inPulsarSourceEnumerator— created on start, closed on close, injected into subscribers that implementRequiresPulsarAdmin.PulsarAdminresource leak by refactoring lifecycle management so the enumerator owns and closes the admin client.lastKnownNamespacesis null after deserialization (checkpoint recovery).Verifying this change
This change added tests and can be verified as follows:
NamespacePatternSubscriberTestwith unit tests covering namespace discovery, topic pattern matching, fallback to last-known namespaces on failure, deserialization safety, and pattern parsing.Significant changes
@Public(Evolving))