Add MemQ support for FlinkSQL 1.18

[jeffxiang]

Summary

This PR makes PSC-Flink compatible with MemQ source for Flink 1.18+ jobs via the following changes.

Changes

memq-client version upgrade to 1.0.2

• Bumps memq.version from 0.2.21 to 1.0.2 in psc/pom.xml, psc-examples/pom.xml, and psc-integration-test/pom.xml.

PscMemqMetadataClient implementation

• New class: PscMemqMetadataClient — implements PscBackendMetadataClient for MemQ. Since MemQ has no dedicated admin client, this uses a transient MemqConsumer to fulfill metadata operations (describeTopicUris, listOffsets, listOffsetsForTimestamps, listOffsetsForConsumerGroup). listTopicRns throws UnsupportedOperationException as MemQ does not support listing all topics.
• New class: PscMemqMetadataClientCreator — annotated with @PscMetadataClientCreatorPlugin(backend = "memq", priority = 1) for auto-discovery via the existing plugin registry.
• New class: PscMetadataClientToMemqConsumerConfigConverter — converts PSC metadata client configuration to MemQ consumer properties.
• MemqTopicUri constructor widened from package-private to public so it can be used by the metadata client.
• PscMetadataClientCreatorManager changed from HashMap to TreeMap for deterministic iteration order of backend creator registrations.

MemQ-Flink metrics support

• New class: MemqSourceReaderMetricsUtil — provides metric filter predicates for MemQ-specific metric names (bytes.consumed.total, notification.records.lag.max) under the memq-consumer-metrics group.
• PscSourceReaderMetrics.registerNumBytesIn and maybeAddPendingRecordsMetric now dispatch to the correct metric filter based on backend type (Kafka vs MemQ) instead of hardcoding the Kafka filter.
• PscSourceReaderMetrics.getBackendFromTags hardened against empty metrics maps (returns "unknown" instead of throwing NoSuchElementException).
• PscSourceReaderMetrics.maybeAddPendingRecordsMetric catches IllegalStateException when the metric is not yet available (MemQ metrics may not be registered until the first poll) and returns null to retry on the next cycle.
• PscMemqConsumer.metrics() now returns live Dropwizard metrics from the underlying MemqConsumer's MetricRegistry, wrapped in a LiveDropwizardMetric adapter that reads current values on each call rather than returning stale snapshots.

Reduce per-record hashcode computation via caching

• TopicUriPartition.hashCode() now caches its result in a transient int cachedHashCode field, invalidated on setTopicUri(). This eliminates redundant String.hashCode() + TopicUri.hashCode() recomputation on every record during the Flink read loop (HashMap lookups in PscPartitionSplitRecords, offset tracking, etc.).

Per-record offset tracking optimization

• PscSourceReaderMetrics.Offset class widened to public with public fields.
• New getOffsetTracker(TopicUriPartition) method allows callers to cache the Offset reference.
• PscPartitionSplitRecords caches the Offset tracker per partition and writes currentOffset directly (currentOffsetTracker.currentOffset = offset) instead of calling metrics.recordCurrentOffset(tp, offset) on every record, eliminating a HashMap lookup per message.

Proper MemQ offset management

• PscMemqConsumer.startOffsets() and endOffsets() now encode raw Kafka notification offsets via kafkaOffsetToComposite() (wrapping them in MemqOffset(kafkaOffset, 0).toLong()). Previously, raw Kafka offsets were returned directly, which caused seekToOffset to misinterpret them — e.g., a raw offset of 2205646 would be decoded as batch offset 4 after the 19-bit right shift in MemqOffset.convertPscOffsetToMemqOffset.
• New test class TestMemqOffsetRoundTrip validates the encoding round-trip and demonstrates the corruption that occurs without it.

Memory optimizations

• PscMetricRegistryManager: replaced SlidingTimeWindowArrayReservoir(1, TimeUnit.MINUTES) with ExponentiallyDecayingReservoir() for histogram metrics. SlidingTimeWindowArrayReservoir retains all samples within the time window, which can grow unbounded under high throughput; ExponentiallyDecayingReservoir uses a fixed-size sample with exponential decay, providing bounded memory usage.

Other fixes

• PscMemqConsumer.close() now calls scheduler.shutdown() to prevent thread leaks.
• PscConsumerMessagesIterable now closes the underlying PscConsumerPollMessageIterator after draining it to asList(), preventing resource leaks (relevant for MemQ batch iterators that hold references to byte buffers).

Test plan

Tested via unit tests, integration tests, and e2e functional testing on Flink clusters.

[vahidhashemian]

Do we need to handle potential null return from this call?

[jeffxiang]

fixed in getBackendFromTags to return unknown if null

[vahidhashemian]

To confirm, is tp a no-op for this function?

[jeffxiang]

yes, it's also a no-op on Kafka side. for some reason when we ported the Kafka connector implementation to PSC, filtering further by tp returned empty results, and it only returned metrics when we didn't filter further by tp. We left the method signature untouched but commented out the tp filtering logic.

[vahidhashemian]

What's the reason for making this part of the public API? Any potential harm?

[jeffxiang]

it needs to be accessed from multiple sub packages. I don't think there is potential harm - it's just a data holder for metrics tracking and doesn't impact any business logic.