[feat][fn] PIP-496: Support the V5 client and topic:// topics in Pulsar Functions and IO

[lhotari]

PIP: PIP-496, proposed in #26698

Motivation

Scalable topics (topic://, PIP-460) can only be produced to and consumed from with the V5 client
(PIP-466). The Pulsar Functions and IO runtime uses the v4 client throughout, so today a function, source or
sink cannot read from or write to a topic:// topic.

It is also a migration blocker. PIP-475 refuses (HTTP 409) to migrate a persistent:// topic to a scalable
topic while any v4 producer or consumer is attached, and every function or connector on the topic attaches
one. A component needs a way to use the V5 client for persistent:// topics too, so that its topics can be
migrated.

This PR implements PIP-496.

Modifications

Client selection. Each component uses one client for its own topics. The runtime picks it with the same
rule as the pulsar-client and pulsar-perf CLIs (#26693):

• A topic:// topic selects the V5 client, and any other domain selects v4.
• An explicit clientApi=V5 makes the V5 client drive persistent:// topics.

The new clientApi setting (V4/V5; unset means "pick from the topics") is added to FunctionConfig,
SourceConfig and SinkConfig, as proto field FunctionDetails.clientApi = 24 (AUTO by default) and as
--client-api on pulsar-admin functions|sources|sinks. ClientApiResolver validates it when a component
is converted, which is when the worker registers or updates it, and again when an instance starts. It rejects:

• mixing topic:// with other domains, or naming a segment:// topic directly;
• V4 with a topic:// topic, and V5 with a non-persistent:// topic;
• with V5: topic patterns, non-Java runtimes, and EFFECTIVELY_ONCE with an output topic (per-segment
  sequence IDs are not monotonic across splits and merges);
• with V5 on persistent:// inputs: any subscription type other than Shared, because a V5 stream consumer
  cannot yet subscribe to a topic that has not been migrated;
• with V5: consumer and producer encryption, message payload processors, consumer properties and
  skipToLatest;
• with V5 and an ordered subscription: maxMessageRetries, deadLetterTopic and timeoutMs, which a stream
  cannot honor.

Instances resolve again when they start, so later versions may relax these rules but must not tighten them.

Consuming (V5PulsarSource). Each input topic gets its own V5 consumer.

• A Shared subscription uses a QueueConsumer. Individual acks, negative acks, the processing timeout, the
  negative-ack delay and the dead letter policy map from the component's settings. Without a configured dead
  letter topic, the v4 default name (<input>-<subscription>-DLQ) is used, so that switching a component to V5
  keeps its dead letter topic.
• Failover, Key_Shared and EFFECTIVELY_ONCE use a StreamConsumer, which delivers in order and splits key
  ranges across the component's instances.
  • A stream acknowledges only cumulatively, so StreamAckTracker acknowledges the completed prefix of the
    received records. Records completed out of order are never skipped.
  • A cumulative acknowledgment of a record completes every record received before it.
  • A stream has no negative acknowledgment, so a failed record fails the instance through its fatal handler,
    from whichever thread calls fail().
• Each start of an instance uses a new consumer name, <tenant>-<namespace>-<name>-<instanceId>-<random>: a
  stream consumer group treats a reused name as a reconnect of the previous member.
• Records are PulsarRecords over the v4 message that the V5 message wraps. Connectors that read the
  schema, the message or the encryption context keep working. The record's topic is the topic:// input,
  not the segment.

Publishing. V5 producers are presented through the v4 Producer/TypedMessageBuilder/MessageId
interfaces, so PulsarSink, ProducerCache, ContextImpl.newOutputMessage and the log appender keep a
single code path; only producer creation branches (V5ProducerFactory). Context.newOutputMessage publishes
to a topic:// topic with the V5 client in every component.

Runtime.

• The thread runtime factory owns one V5 client. It is created on first use, shared by the instances, and
  closed with the factory.
• InstanceUtils.createPulsarClientV5Builder applies the service URL, authentication, TLS and memory limit
  of the v4 client. The TLS policy is set only when TLS is enabled, and hostname verification is always
  passed explicitly, because the TLS policy's default differs from the v4 client's.
• Without a configured memory limit, the V5 client keeps its 64 MiB default. The V5 producer has no
  pending-message limit, so the memory limit is what bounds it.
• The V5 client implementation is looked up once under the runtime's class loader, so that the first V5 call
  from user code in the process and Kubernetes runtimes cannot fail the JVM-wide lookup.

User API (PIP-496): one default method, BaseContext.getPulsarClientV5(), which returns the runtime's
shared V5 client.

Packaging and worker.

• pulsar-functions-api depends on pulsar-client-api-v5, so the V5 API is in java-instance.jar. The
  OpenTelemetry API is kept out of that jar, as with the v4 API.
• pulsar-io-core now depends on the V5 API, so pulsar-client-api-v5, pulsar-client-v5,
  pulsar-tls-factory-api and pulsar-http-client-api are added to the NAR platform exclusions: the runtime
  provides them.
• cleanupSubscription deletes a topic:// input's subscription through the scalable topics admin API.
• Triggering a function with a topic:// topic is rejected with 400.

V5 client fix. The commit "[fix][client] Retry a V5 stream consumer's initial subscribe with the
current assignment" fixes a bug that this work exposed. When two stream consumers join a subscription at the
same moment, the first one's initial subscribe retried with a stale assignment and never completed. It can be
split into its own PR.

Not supported with the V5 client in this PR (each is rejected with an explicit error):

• consumer and producer encryption;
• message payload processors and consumer properties;
• skipToLatest;
• SinkContext seek, pause and resume (they throw UnsupportedOperationException at call time);
• per-message schemas on one producer.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Unit tests:
  • ClientApiResolverTest: the selection matrix and every rejection.
  • FunctionConfigUtilsTest, SinkConfigUtilsTest and SourceConfigUtilsTest: round-trip, update and
    rejection of clientApi.
  • CmdFunctionsTest: --client-api.
  • V5InteropTest, V5ProducerAdapterTest, V5ProducerFactoryTest and LazyPulsarClientV5Test.
  • StreamAckTrackerTest, including out-of-order completion and cumulative acknowledgment.
  • ContextImplTest: getPulsarClientV5(), publishing to topic:// from v4 and V5 components, and the
    rejection of seek/pause/resume.
• End-to-end, against a TLS and TLS-authentication broker with the function worker (thread runtime):
  • PulsarSourceV5E2ETest:
    • a source publishes to a topic:// topic;
    • with clientApi=V5 on a persistent:// output, the publisher carries the V5 marker that PIP-475
      migration checks;
    • V4 with a topic:// output is rejected.
  • PulsarFunctionV5E2ETest:
    • A Shared function on topic:// in and out; every input message is acknowledged, and deleting the
      function removes the subscription.
    • A Key_Shared function with parallelism 2; each key's messages arrive in order. Before the client fix,
      one instance never finished subscribing.
    • A clientApi=V5 Shared function on persistent:// topics; its consumer carries the V5 marker, and
      migrateToScalable(input, force=false) succeeds while the function is running.
• Mutation checks:
  • forcing the runtime back to the v4 client fails both positive source tests;
  • dropping the scalable-topic branch of the subscription cleanup fails the Shared function test.
• Regression:
  • the existing PulsarFunctionE2ETest, PulsarSinkE2ETest, PulsarSourceE2ETest,
    PulsarBatchSourceE2ETest and PulsarFunctionTlsTest pass unchanged;
  • all tests of the functions worker, runtime, runtime-all, instance and utils modules pass;
  • ./gradlew sanityCheck passes.
• Not verified yet:
  • the process and Kubernetes runtimes, where user code loads java-instance.jar in its own class loader;
  • the batch source with V5 on a live broker;
  • the fatal-handler path of a failed stream record. It has no automated test, because building a real V5
    message in a unit test needs package-private client classes.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
  pulsar-functions-api now depends on pulsar-client-api-v5, and pulsar-functions-instance on
  pulsar-client-v5. Both are Pulsar modules, with no new third-party libraries.
• The public API
  Added BaseContext.getPulsarClientV5(), a default method, and a clientApi field to FunctionConfig,
  SourceConfig and SinkConfig. Context.newOutputMessage now publishes to topic:// topics with the V5
  client.
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
  The function, source and sink configs accept clientApi. Trigger rejects topic:// topics.
• The admin CLI options
  --client-api on pulsar-admin functions|sources|sinks create|update|localrun.
• The metrics
• Anything that affects deployment
  java-instance.jar now contains the V5 client API. NARs exclude the V5 client modules that the runtime
  provides.