You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/04/20 20:19:51 UTC
(pulsar) branch master updated: [feat] PIP-466: Add V5 client API for scalable topics (#25489)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 6ce15e13cc3 [feat] PIP-466: Add V5 client API for scalable topics (#25489)
6ce15e13cc3 is described below
commit 6ce15e13cc37bb1cd9ceae1814e659c420f9550b
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Apr 20 13:19:45 2026 -0700
[feat] PIP-466: Add V5 client API for scalable topics (#25489)
---
pulsar-client-api-v5/build.gradle.kts | 27 ++
.../apache/pulsar/client/api/v5/Checkpoint.java | 95 ++++
.../pulsar/client/api/v5/CheckpointConsumer.java | 123 +++++
.../client/api/v5/CheckpointConsumerBuilder.java | 113 +++++
.../org/apache/pulsar/client/api/v5/Message.java | 123 +++++
.../pulsar/client/api/v5/MessageBuilder.java | 39 ++
.../org/apache/pulsar/client/api/v5/MessageId.java | 68 +++
.../pulsar/client/api/v5/MessageMetadata.java | 120 +++++
.../org/apache/pulsar/client/api/v5/Messages.java | 41 ++
.../org/apache/pulsar/client/api/v5/Producer.java | 80 +++
.../pulsar/client/api/v5/ProducerBuilder.java | 170 +++++++
.../apache/pulsar/client/api/v5/PulsarClient.java | 130 +++++
.../pulsar/client/api/v5/PulsarClientBuilder.java | 156 ++++++
.../client/api/v5/PulsarClientException.java | 161 ++++++
.../apache/pulsar/client/api/v5/QueueConsumer.java | 117 +++++
.../pulsar/client/api/v5/QueueConsumerBuilder.java | 239 +++++++++
.../pulsar/client/api/v5/StreamConsumer.java | 121 +++++
.../client/api/v5/StreamConsumerBuilder.java | 166 +++++++
.../apache/pulsar/client/api/v5/Transaction.java | 71 +++
.../api/v5/async/AsyncCheckpointConsumer.java | 88 ++++
.../client/api/v5/async/AsyncMessageBuilder.java | 42 ++
.../pulsar/client/api/v5/async/AsyncProducer.java | 57 +++
.../client/api/v5/async/AsyncQueueConsumer.java | 88 ++++
.../client/api/v5/async/AsyncStreamConsumer.java | 89 ++++
.../client/api/v5/async/AsyncTransaction.java | 48 ++
.../pulsar/client/api/v5/async/package-info.java | 28 ++
.../pulsar/client/api/v5/auth/Authentication.java | 70 +++
.../client/api/v5/auth/AuthenticationData.java | 99 ++++
.../client/api/v5/auth/AuthenticationFactory.java | 93 ++++
.../client/api/v5/auth/CryptoFailureAction.java | 41 ++
.../pulsar/client/api/v5/auth/CryptoKeyReader.java | 45 ++
.../client/api/v5/auth/EncryptionKeyInfo.java | 37 ++
.../pulsar/client/api/v5/auth/package-info.java | 28 ++
.../pulsar/client/api/v5/config/BackoffPolicy.java | 79 +++
.../client/api/v5/config/BatchingPolicy.java | 87 ++++
.../client/api/v5/config/ChunkingPolicy.java | 31 ++
.../client/api/v5/config/CompressionPolicy.java | 46 ++
.../client/api/v5/config/CompressionType.java | 52 ++
.../client/api/v5/config/ConnectionPolicy.java | 217 +++++++++
.../client/api/v5/config/DeadLetterPolicy.java | 53 ++
.../client/api/v5/config/EncryptionPolicy.java | 72 +++
.../pulsar/client/api/v5/config/MemorySize.java | 101 ++++
.../client/api/v5/config/ProducerAccessMode.java | 47 ++
.../pulsar/client/api/v5/config/ProxyProtocol.java | 26 +
.../api/v5/config/SubscriptionInitialPosition.java | 35 ++
.../pulsar/client/api/v5/config/TlsPolicy.java | 67 +++
.../client/api/v5/config/TransactionPolicy.java | 30 ++
.../pulsar/client/api/v5/config/package-info.java | 29 ++
.../api/v5/internal/PulsarClientProvider.java | 123 +++++
.../client/api/v5/internal/package-info.java | 23 +
.../apache/pulsar/client/api/v5/package-info.java | 46 ++
.../apache/pulsar/client/api/v5/schema/Schema.java | 207 ++++++++
.../pulsar/client/api/v5/schema/SchemaInfo.java | 55 +++
.../pulsar/client/api/v5/schema/SchemaType.java | 42 ++
.../pulsar/client/api/v5/schema/package-info.java | 27 ++
.../org/apache/pulsar/client/api/v5/Examples.java | 538 +++++++++++++++++++++
settings.gradle.kts | 2 +
57 files changed, 5048 insertions(+)
diff --git a/pulsar-client-api-v5/build.gradle.kts b/pulsar-client-api-v5/build.gradle.kts
new file mode 100644
index 00000000000..fa777518491
--- /dev/null
+++ b/pulsar-client-api-v5/build.gradle.kts
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+plugins {
+ id("pulsar.java-conventions")
+}
+
+dependencies {
+ compileOnly(libs.protobuf.java)
+ compileOnly(libs.opentelemetry.api)
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Checkpoint.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Checkpoint.java
new file mode 100644
index 00000000000..6b24fe8943d
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Checkpoint.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.io.IOException;
+import java.time.Instant;
+import org.apache.pulsar.client.api.v5.internal.PulsarClientProvider;
+
+/**
+ * An opaque, serializable position vector representing a consistent point across all
+ * internal hash-range segments of a topic.
+ *
+ * <p>Checkpoints are created via {@link CheckpointConsumer#checkpoint()} and can be
+ * serialized for external storage (e.g. Flink state, S3) using {@link #toByteArray()}.
+ *
+ * <p>This is the sole position type used with {@link CheckpointConsumer} — for initial
+ * positioning use the static factories {@link #earliest()}, {@link #latest()},
+ * {@link #atTimestamp(Instant)}, or {@link #fromByteArray(byte[])} to restore from
+ * a previously saved checkpoint.
+ */
+public interface Checkpoint {
+
+ /**
+ * Serialize this checkpoint for external storage.
+ *
+ * @return a serializable byte representation of this checkpoint that can be restored
+ * via {@link #fromByteArray(byte[])}
+ */
+ byte[] toByteArray();
+
+ /**
+ * The time at which this checkpoint was created.
+ *
+ * @return the creation timestamp of this checkpoint as an {@link Instant}
+ */
+ Instant creationTime();
+
+ // --- Static factories ---
+
+ /**
+ * A sentinel checkpoint representing the beginning of the topic (oldest available data).
+ *
+ * @return a sentinel {@link Checkpoint} representing the earliest position in the topic
+ */
+ static Checkpoint earliest() {
+ return PulsarClientProvider.get().earliestCheckpoint();
+ }
+
+ /**
+ * A sentinel checkpoint representing the end of the topic (next message to be published).
+ *
+ * @return a sentinel {@link Checkpoint} representing the latest position in the topic
+ */
+ static Checkpoint latest() {
+ return PulsarClientProvider.get().latestCheckpoint();
+ }
+
+ /**
+ * A checkpoint that positions at the first message published at or after the given timestamp.
+ *
+ * @param timestamp the timestamp to position at
+ * @return a {@link Checkpoint} that will start consuming from the first message at or after
+ * the given timestamp
+ */
+ static Checkpoint atTimestamp(Instant timestamp) {
+ return PulsarClientProvider.get().checkpointAtTimestamp(timestamp);
+ }
+
+ /**
+ * Deserialize a checkpoint from a byte array previously obtained via {@link #toByteArray()}.
+ *
+ * @param data the byte array previously obtained from {@link #toByteArray()}
+ * @return the deserialized {@link Checkpoint}
+ * @throws IOException if the byte array is malformed or cannot be deserialized
+ */
+ static Checkpoint fromByteArray(byte[] data) throws IOException {
+ return PulsarClientProvider.get().checkpointFromBytes(data);
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/CheckpointConsumer.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/CheckpointConsumer.java
new file mode 100644
index 00000000000..0d278fb3eb5
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/CheckpointConsumer.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.io.Closeable;
+import java.time.Duration;
+import org.apache.pulsar.client.api.v5.async.AsyncCheckpointConsumer;
+
+/**
+ * An unmanaged consumer designed for connector frameworks (Flink, Spark, etc.).
+ *
+ * <p>Unlike {@link StreamConsumer} and {@link QueueConsumer}, this consumer has no
+ * broker-managed subscription — position tracking is entirely external. The connector
+ * framework stores checkpoints in its own state backend and uses them to restore
+ * on failure.
+ *
+ * <p>Internally, the consumer reads from all hash-range segments of a topic.
+ * {@link #checkpoint()} creates an atomic snapshot of positions across all segments,
+ * returned as an opaque {@link Checkpoint} that can be serialized and stored externally.
+ *
+ * <p>This interface provides synchronous (blocking) operations. For non-blocking
+ * usage, obtain an {@link AsyncCheckpointConsumer} via {@link #async()}.
+ *
+ * @param <T> the type of message values
+ */
+public interface CheckpointConsumer<T> extends Closeable {
+
+ /**
+ * The topic this consumer reads from.
+ *
+ * @return the fully qualified topic name
+ */
+ String topic();
+
+ // --- Receive ---
+
+ /**
+ * Receive a single message, blocking indefinitely.
+ *
+ * @return the received {@link Message}
+ * @throws PulsarClientException if the consumer is closed or a connection error occurs
+ */
+ Message<T> receive() throws PulsarClientException;
+
+ /**
+ * Receive a single message, blocking up to the given timeout.
+ * Returns {@code null} if the timeout elapses without a message.
+ *
+ * @param timeout the maximum time to wait for a message
+ * @return the received {@link Message}, or {@code null} if the timeout elapses
+ * @throws PulsarClientException if the consumer is closed or a connection error occurs
+ */
+ Message<T> receive(Duration timeout) throws PulsarClientException;
+
+ /**
+ * Receive a batch of messages, blocking up to the given timeout.
+ *
+ * @param maxMessages the maximum number of messages to return
+ * @param timeout the maximum time to wait for messages
+ * @return the received {@link Messages} batch
+ * @throws PulsarClientException if the consumer is closed or a connection error occurs
+ */
+ Messages<T> receiveMulti(int maxMessages, Duration timeout) throws PulsarClientException;
+
+ // --- Checkpoint ---
+
+ /**
+ * Create a consistent checkpoint — an atomic snapshot of positions across all
+ * internal hash-range segments.
+ *
+ * <p>The returned {@link Checkpoint} can be serialized via {@link Checkpoint#toByteArray()}
+ * and stored in the connector framework's state backend.
+ *
+ * @return an opaque {@link Checkpoint} representing the current read positions
+ */
+ Checkpoint checkpoint();
+
+ // --- Seek ---
+
+ /**
+ * Seek to a previously saved checkpoint, or to a sentinel position such as
+ * {@link Checkpoint#earliest()} or {@link Checkpoint#latest()}.
+ *
+ * @param checkpoint the checkpoint to seek to
+ * @throws PulsarClientException if the seek fails or a connection error occurs
+ */
+ void seek(Checkpoint checkpoint) throws PulsarClientException;
+
+ // --- Async ---
+
+ /**
+ * Return the asynchronous view of this consumer.
+ *
+ * @return the {@link AsyncCheckpointConsumer} counterpart of this consumer
+ */
+ AsyncCheckpointConsumer<T> async();
+
+ // --- Lifecycle ---
+
+ /**
+ * Close the consumer and release all resources.
+ *
+ * @throws PulsarClientException if an error occurs while closing the consumer
+ */
+ @Override
+ void close() throws PulsarClientException;
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/CheckpointConsumerBuilder.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/CheckpointConsumerBuilder.java
new file mode 100644
index 00000000000..615473a382c
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/CheckpointConsumerBuilder.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.config.EncryptionPolicy;
+
+/**
+ * Builder for configuring and creating a {@link CheckpointConsumer}.
+ *
+ * <p>Since this is an unmanaged consumer (no subscription), the terminal method is
+ * {@link #create()} rather than {@code subscribe()}.
+ *
+ * @param <T> the type of message values the consumer will receive
+ */
+public interface CheckpointConsumerBuilder<T> {
+
+ /**
+ * Create the checkpoint consumer, blocking until it is ready.
+ *
+ * @return the created {@link CheckpointConsumer}
+ * @throws PulsarClientException if the creation fails or a connection error occurs
+ */
+ CheckpointConsumer<T> create() throws PulsarClientException;
+
+ /**
+ * Create the checkpoint consumer asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes with the created {@link CheckpointConsumer}
+ */
+ CompletableFuture<CheckpointConsumer<T>> createAsync();
+
+ // --- Required ---
+
+ /**
+ * The topic to consume from.
+ *
+ * @param topicName the topic name
+ * @return this builder instance for chaining
+ */
+ CheckpointConsumerBuilder<T> topic(String topicName);
+
+ // --- Start position ---
+
+ /**
+ * Set the initial position for this consumer.
+ *
+ * <p>Use {@link Checkpoint#earliest()}, {@link Checkpoint#latest()},
+ * {@link Checkpoint#atTimestamp}, or {@link Checkpoint#fromByteArray} to
+ * create the appropriate starting position.
+ *
+ * <p>Defaults to {@link Checkpoint#latest()} if not specified.
+ *
+ * @param checkpoint the checkpoint representing the desired start position
+ * @return this builder instance for chaining
+ */
+ CheckpointConsumerBuilder<T> startPosition(Checkpoint checkpoint);
+
+ // --- Optional ---
+
+ /**
+ * A custom name for this consumer instance.
+ *
+ * @param name the consumer name
+ * @return this builder instance for chaining
+ */
+ CheckpointConsumerBuilder<T> consumerName(String name);
+
+ /**
+ * Configure end-to-end message encryption for decryption.
+ *
+ * @param policy the encryption policy to use
+ * @return this builder instance for chaining
+ * @see EncryptionPolicy#forConsumer
+ */
+ CheckpointConsumerBuilder<T> encryptionPolicy(EncryptionPolicy policy);
+
+ // --- Metadata ---
+
+ /**
+ * Add a single property to the consumer metadata.
+ *
+ * @param key the property key
+ * @param value the property value
+ * @return this builder instance for chaining
+ */
+ CheckpointConsumerBuilder<T> property(String key, String value);
+
+ /**
+ * Add multiple properties to the consumer metadata.
+ *
+ * @param properties the properties to add
+ * @return this builder instance for chaining
+ */
+ CheckpointConsumerBuilder<T> properties(Map<String, String> properties);
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Message.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Message.java
new file mode 100644
index 00000000000..e7bfad0d49a
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Message.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * An immutable message received from a Pulsar topic.
+ *
+ * @param <T> the type of the deserialized message value
+ */
+public interface Message<T> {
+
+ /**
+ * The deserialized value of the message according to the schema.
+ *
+ * @return the deserialized message value
+ */
+ T value();
+
+ /**
+ * The raw bytes of the message payload.
+ *
+ * @return the raw payload as a byte array
+ */
+ byte[] data();
+
+ /**
+ * The unique identifier of this message within the topic.
+ *
+ * @return the {@link MessageId} of this message
+ */
+ MessageId id();
+
+ /**
+ * The message key, used for per-key ordering.
+ *
+ * @return an {@link Optional} containing the message key, or empty if no key was set
+ */
+ Optional<String> key();
+
+ /**
+ * Application-defined properties attached to the message.
+ *
+ * @return an unmodifiable map of property key-value pairs
+ */
+ Map<String, String> properties();
+
+ /**
+ * The timestamp when the message was published by the broker.
+ *
+ * @return the publish timestamp as an {@link Instant}
+ */
+ Instant publishTime();
+
+ /**
+ * The event time set by the producer, if any.
+ *
+ * @return an {@link Optional} containing the event time, or empty if not set by the producer
+ */
+ Optional<Instant> eventTime();
+
+ /**
+ * The producer-assigned sequence ID for deduplication.
+ *
+ * @return the sequence ID of this message
+ */
+ long sequenceId();
+
+ /**
+ * The name of the producer that published this message.
+ *
+ * @return an {@link Optional} containing the producer name, or empty if not available
+ */
+ Optional<String> producerName();
+
+ /**
+ * The topic this message was published to.
+ *
+ * @return the fully qualified topic name
+ */
+ String topic();
+
+ /**
+ * The number of times the broker has redelivered this message.
+ *
+ * @return the redelivery count, starting at 0 for the first delivery
+ */
+ int redeliveryCount();
+
+ /**
+ * The uncompressed size of the message payload in bytes.
+ *
+ * @return the payload size in bytes
+ */
+ int size();
+
+ /**
+ * The cluster from which this message was replicated, if applicable.
+ *
+ * @return an {@link Optional} containing the source cluster name, or empty if the message
+ * is not replicated
+ */
+ Optional<String> replicatedFrom();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/MessageBuilder.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/MessageBuilder.java
new file mode 100644
index 00000000000..e657edf788a
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/MessageBuilder.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+/**
+ * Synchronous message builder, obtained from {@link Producer#newMessage()}.
+ *
+ * <p>Inherits all metadata setters from {@link MessageMetadata} and adds a
+ * blocking {@link #send()} terminal operation.
+ *
+ * @param <T> the type of the message value
+ */
+public interface MessageBuilder<T> extends MessageMetadata<T, MessageBuilder<T>> {
+
+ /**
+ * Send the message synchronously and return its message ID.
+ *
+ * @return the {@link MessageId} assigned to the published message by the broker
+ * @throws PulsarClientException if the message could not be sent (e.g., connection failure,
+ * send timeout, or topic authorization error)
+ */
+ MessageId send() throws PulsarClientException;
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/MessageId.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/MessageId.java
new file mode 100644
index 00000000000..fc539e0266d
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/MessageId.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.io.IOException;
+import org.apache.pulsar.client.api.v5.internal.PulsarClientProvider;
+
+/**
+ * Opaque, immutable identifier for a message within a topic.
+ *
+ * <p>No internal structure (ledger ID, entry ID, partition index) is exposed.
+ * Message IDs can be serialized to bytes for external storage and restored later.
+ */
+public interface MessageId extends Comparable<MessageId> {
+
+ /**
+ * Serialize this message ID to a byte array for external storage.
+ *
+ * @return a byte array representation of this message ID that can be restored
+ * via {@link #fromByteArray(byte[])}
+ */
+ byte[] toByteArray();
+
+ /**
+ * Deserialize a message ID from bytes previously produced by {@link #toByteArray()}.
+ *
+ * @param data the byte array previously obtained from {@link #toByteArray()}
+ * @return the deserialized {@link MessageId}
+ * @throws IOException if the byte array is malformed or cannot be deserialized
+ */
+ static MessageId fromByteArray(byte[] data) throws IOException {
+ return PulsarClientProvider.get().messageIdFromBytes(data);
+ }
+
+ /**
+ * Sentinel representing the oldest available message in the topic.
+ *
+ * @return a sentinel {@link MessageId} representing the earliest position in the topic
+ */
+ static MessageId earliest() {
+ return PulsarClientProvider.get().earliestMessageId();
+ }
+
+ /**
+ * Sentinel representing the next message to be published (i.e., the end of the topic).
+ *
+ * @return a sentinel {@link MessageId} representing the latest position in the topic
+ */
+ static MessageId latest() {
+ return PulsarClientProvider.get().latestMessageId();
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/MessageMetadata.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/MessageMetadata.java
new file mode 100644
index 00000000000..2de48c59ce9
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/MessageMetadata.java
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Common message metadata that can be set on any outgoing message.
+ *
+ * <p>This is the shared base for {@link MessageBuilder} (sync) and
+ * {@link org.apache.pulsar.client.api.v5.async.AsyncMessageBuilder AsyncMessageBuilder} (async).
+ * The self-referential type parameter {@code BuilderT} enables fluent chaining
+ * on both subtypes.
+ *
+ * @param <T> the type of the message value
+ * @param <BuilderT> the concrete builder type (for fluent returns)
+ */
+public interface MessageMetadata<T, BuilderT extends MessageMetadata<T, BuilderT>> {
+
+ /**
+ * Set the message value.
+ *
+ * @param value the message payload to be serialized using the producer's schema
+ * @return this builder instance for chaining
+ */
+ BuilderT value(T value);
+
+ /**
+ * Set the message key. Messages with the same key are guaranteed to be delivered
+ * in order to stream consumers. Queue consumers may use the key for routing.
+ *
+ * @param key the message key used for ordering and routing
+ * @return this builder instance for chaining
+ */
+ BuilderT key(String key);
+
+ /**
+ * Associate this message with a transaction.
+ *
+ * @param txn the transaction to associate with this message
+ * @return this builder instance for chaining
+ */
+ BuilderT transaction(Transaction txn);
+
+ /**
+ * Add a single property to the message.
+ *
+ * @param name the property key
+ * @param value the property value
+ * @return this builder instance for chaining
+ */
+ BuilderT property(String name, String value);
+
+ /**
+ * Add multiple properties to the message.
+ *
+ * @param properties a map of property key-value pairs to attach to the message
+ * @return this builder instance for chaining
+ */
+ BuilderT properties(Map<String, String> properties);
+
+ /**
+ * Set the event time of the message.
+ *
+ * @param eventTime the application-defined event time for the message
+ * @return this builder instance for chaining
+ */
+ BuilderT eventTime(Instant eventTime);
+
+ /**
+ * Set the sequence ID for producer deduplication.
+ *
+ * @param sequenceId the sequence ID to assign to the message for deduplication purposes
+ * @return this builder instance for chaining
+ */
+ BuilderT sequenceId(long sequenceId);
+
+ /**
+ * Request delayed delivery: the message becomes visible to consumers after the given delay.
+ *
+ * @param delay the duration to wait before the message becomes visible to consumers
+ * @return this builder instance for chaining
+ */
+ BuilderT deliverAfter(Duration delay);
+
+ /**
+ * Request delayed delivery: the message becomes visible to consumers at the given time.
+ *
+ * @param timestamp the absolute time at which the message becomes visible to consumers
+ * @return this builder instance for chaining
+ */
+ BuilderT deliverAt(Instant timestamp);
+
+ /**
+ * Restrict geo-replication to the specified clusters only.
+ *
+ * @param clusters the list of cluster names to which this message should be replicated
+ * @return this builder instance for chaining
+ */
+ BuilderT replicationClusters(List<String> clusters);
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Messages.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Messages.java
new file mode 100644
index 00000000000..4fcb5df17e3
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Messages.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+/**
+ * A batch of messages received from a consumer.
+ *
+ * @param <T> the type of the deserialized message value
+ */
+public interface Messages<T> extends Iterable<Message<T>> {
+
+ /**
+ * The number of messages in this batch.
+ *
+ * @return the number of messages in this batch
+ */
+ int count();
+
+ /**
+ * The last message id in this batch.
+ *
+ * @return the {@link MessageId} of the last message in this batch
+ */
+ MessageId lastId();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Producer.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Producer.java
new file mode 100644
index 00000000000..5a49eac7d71
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Producer.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import org.apache.pulsar.client.api.v5.async.AsyncProducer;
+
+/**
+ * A producer publishes messages to a Pulsar topic.
+ *
+ * <p>This interface provides synchronous (blocking) operations. For non-blocking
+ * usage, obtain an {@link AsyncProducer} via {@link #async()}.
+ *
+ * @param <T> the type of message values this producer sends
+ */
+public interface Producer<T> extends AutoCloseable {
+
+ /**
+ * The topic this producer is attached to.
+ *
+ * @return the fully qualified topic name (e.g. {@code topic://tenant/namespace/my-topic})
+ */
+ String topic();
+
+ /**
+ * The name of this producer (system-assigned or user-specified via
+ * {@link ProducerBuilder#producerName(String)}).
+ *
+ * @return the producer name, never {@code null}
+ */
+ String producerName();
+
+ /**
+ * Create a message builder for advanced message construction (key, properties, etc.).
+ * Use {@link MessageBuilder#send()} as the terminal operation.
+ *
+ * @return a new {@link MessageBuilder} instance bound to this producer
+ */
+ MessageBuilder<T> newMessage();
+
+ /**
+ * The last sequence ID published by this producer. Used for deduplication tracking.
+ * Returns {@code -1} if no message has been published yet.
+ *
+ * @return the last published sequence ID, or {@code -1} if none
+ */
+ long lastSequenceId();
+
+ /**
+ * Return the asynchronous view of this producer. The returned object shares the same
+ * underlying connection and resources.
+ *
+ * @return the {@link AsyncProducer} counterpart of this producer
+ */
+ AsyncProducer<T> async();
+
+ /**
+ * Close this producer and release all associated resources. Pending send operations
+ * are completed before the producer is closed.
+ *
+ * @throws PulsarClientException if an error occurs while closing
+ */
+ @Override
+ void close() throws PulsarClientException;
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/ProducerBuilder.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/ProducerBuilder.java
new file mode 100644
index 00000000000..65800756c19
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/ProducerBuilder.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.config.BatchingPolicy;
+import org.apache.pulsar.client.api.v5.config.ChunkingPolicy;
+import org.apache.pulsar.client.api.v5.config.CompressionPolicy;
+import org.apache.pulsar.client.api.v5.config.EncryptionPolicy;
+import org.apache.pulsar.client.api.v5.config.ProducerAccessMode;
+
+/**
+ * Builder for configuring and creating a {@link Producer}.
+ *
+ * @param <T> the type of message values the producer will send
+ */
+public interface ProducerBuilder<T> {
+
+ /**
+ * Create the producer, blocking until it is ready.
+ *
+ * @return the configured {@link Producer} instance
+ * @throws PulsarClientException if the producer cannot be created (e.g. topic does not exist,
+ * authorization failure, or connection error)
+ */
+ Producer<T> create() throws PulsarClientException;
+
+ /**
+ * Create the producer asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes with the configured {@link Producer},
+ * or completes exceptionally with {@link PulsarClientException} on failure
+ */
+ CompletableFuture<Producer<T>> createAsync();
+
+ // --- Required ---
+
+ /**
+ * The topic to produce to. This is required and must be set before calling {@link #create()}.
+ *
+ * @param topicName the fully qualified topic name (e.g. {@code topic://tenant/namespace/my-topic})
+ * @return this builder instance for chaining
+ */
+ ProducerBuilder<T> topic(String topicName);
+
+ // --- Optional ---
+
+ /**
+ * Set a custom producer name. If not set, the broker assigns a unique name.
+ * The producer name is used for message deduplication and appears in broker logs.
+ *
+ * @param producerName the producer name
+ * @return this builder instance for chaining
+ */
+ ProducerBuilder<T> producerName(String producerName);
+
+ /**
+ * Access mode for this producer on the topic.
+ *
+ * @param accessMode the access mode (e.g. {@link ProducerAccessMode#SHARED},
+ * {@link ProducerAccessMode#EXCLUSIVE})
+ * @return this builder instance for chaining
+ */
+ ProducerBuilder<T> accessMode(ProducerAccessMode accessMode);
+
+ /**
+ * Timeout for a send operation. If the message is not acknowledged by the broker within
+ * this duration, the send future completes exceptionally. A value of {@link Duration#ZERO}
+ * disables the timeout.
+ *
+ * @param timeout the send timeout duration
+ * @return this builder instance for chaining
+ */
+ ProducerBuilder<T> sendTimeout(Duration timeout);
+
+ /**
+ * Whether the producer should block when the pending message queue is full,
+ * rather than failing immediately. Default is {@code true}.
+ *
+ * @param blockIfQueueFull {@code true} to block, {@code false} to fail immediately
+ * @return this builder instance for chaining
+ */
+ ProducerBuilder<T> blockIfQueueFull(boolean blockIfQueueFull);
+
+ /**
+ * Configure compression for message payloads.
+ *
+ * @param policy the compression policy
+ * @return this builder instance for chaining
+ * @see CompressionPolicy#of(org.apache.pulsar.client.api.v5.config.CompressionType)
+ * @see CompressionPolicy#disabled()
+ */
+ ProducerBuilder<T> compressionPolicy(CompressionPolicy policy);
+
+ /**
+ * Configure message batching. When enabled, the producer groups multiple messages
+ * into a single broker request to improve throughput.
+ *
+ * @param policy the batching policy
+ * @return this builder instance for chaining
+ * @see BatchingPolicy#ofDefault()
+ * @see BatchingPolicy#ofDisabled()
+ * @see BatchingPolicy#of(Duration, int, int)
+ */
+ ProducerBuilder<T> batchingPolicy(BatchingPolicy policy);
+
+ /**
+ * Enable chunking for large messages that exceed the broker's max message size.
+ *
+ * @param policy the chunking policy
+ * @return this builder instance for chaining
+ */
+ ProducerBuilder<T> chunkingPolicy(ChunkingPolicy policy);
+
+ /**
+ * Configure end-to-end message encryption.
+ *
+ * @param policy the encryption policy for producing encrypted messages
+ * @return this builder instance for chaining
+ * @see EncryptionPolicy#forProducer(org.apache.pulsar.client.api.v5.auth.CryptoKeyReader, String...)
+ */
+ ProducerBuilder<T> encryptionPolicy(EncryptionPolicy policy);
+
+ /**
+ * Set the initial sequence ID for producer message deduplication. Subsequent messages
+ * are assigned incrementing sequence IDs starting from this value.
+ *
+ * @param initialSequenceId the starting sequence ID
+ * @return this builder instance for chaining
+ */
+ ProducerBuilder<T> initialSequenceId(long initialSequenceId);
+
+ // --- Metadata ---
+
+ /**
+ * Add a single property to the producer metadata. Properties are sent to the broker
+ * and can be used for filtering and identification.
+ *
+ * @param key the property key
+ * @param value the property value
+ * @return this builder instance for chaining
+ */
+ ProducerBuilder<T> property(String key, String value);
+
+ /**
+ * Add multiple properties to the producer metadata.
+ *
+ * @param properties a map of property key-value pairs
+ * @return this builder instance for chaining
+ */
+ ProducerBuilder<T> properties(Map<String, String> properties);
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/PulsarClient.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/PulsarClient.java
new file mode 100644
index 00000000000..51bf5748c6d
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/PulsarClient.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.internal.PulsarClientProvider;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+
+/**
+ * Entry point for the Pulsar client. Provides factory methods for creating producers,
+ * consumers, and transactions.
+ *
+ * <p>Instances are created via {@link #builder()}.
+ *
+ * <p>A {@code PulsarClient} manages internal resources such as connections, threads,
+ * and memory buffers. It must be closed when no longer needed.
+ */
+public interface PulsarClient extends AutoCloseable {
+
+ /**
+ * Create a new client builder.
+ *
+ * @return a new {@link PulsarClientBuilder} for configuring the client
+ */
+ static PulsarClientBuilder builder() {
+ return PulsarClientProvider.get().newClientBuilder();
+ }
+
+ /**
+ * Create a producer builder with a specific schema.
+ *
+ * @param <T> the message value type
+ * @param schema the schema used for serialization/deserialization
+ * @return a new {@link ProducerBuilder} for configuring the producer
+ */
+ <T> ProducerBuilder<T> newProducer(Schema<T> schema);
+
+ /**
+ * Create a stream consumer builder with a specific schema.
+ *
+ * @param <T> the message value type
+ * @param schema the schema used for serialization/deserialization
+ * @return a new {@link StreamConsumerBuilder} for configuring the stream consumer
+ */
+ <T> StreamConsumerBuilder<T> newStreamConsumer(Schema<T> schema);
+
+ /**
+ * Create a queue consumer builder with a specific schema.
+ *
+ * @param <T> the message value type
+ * @param schema the schema used for serialization/deserialization
+ * @return a new {@link QueueConsumerBuilder} for configuring the queue consumer
+ */
+ <T> QueueConsumerBuilder<T> newQueueConsumer(Schema<T> schema);
+
+ /**
+ * Create a checkpoint consumer builder with a specific schema.
+ *
+ * <p>Checkpoint consumers are unmanaged — position tracking is external.
+ * Designed for connector frameworks (Flink, Spark) that manage their own state.
+ *
+ * @param <T> the message value type
+ * @param schema the schema used for serialization/deserialization
+ * @return a new {@link CheckpointConsumerBuilder} for configuring the checkpoint consumer
+ */
+ <T> CheckpointConsumerBuilder<T> newCheckpointConsumer(Schema<T> schema);
+
+ // --- Transactions ---
+
+ /**
+ * Create a new transaction, blocking until it is ready. The transaction timeout is taken
+ * from the client-wide {@link org.apache.pulsar.client.api.v5.config.TransactionPolicy}
+ * configured on {@link PulsarClientBuilder#transactionPolicy}.
+ *
+ * @return a new {@link Transaction} in the {@link Transaction.State#OPEN} state
+ * @throws PulsarClientException if the transaction cannot be created (e.g., transaction
+ * coordinator unavailable or the client is closed)
+ */
+ Transaction newTransaction() throws PulsarClientException;
+
+ /**
+ * Asynchronous counterpart of {@link #newTransaction()}.
+ *
+ * @return a {@link CompletableFuture} that completes with a new {@link Transaction} in the
+ * {@link Transaction.State#OPEN} state, or completes exceptionally with
+ * {@link PulsarClientException} on failure
+ */
+ CompletableFuture<Transaction> newTransactionAsync();
+
+ // --- Lifecycle ---
+
+ /**
+ * Close the client and release all resources, waiting for pending operations to complete.
+ *
+ * @throws PulsarClientException if an error occurs while closing the client
+ */
+ @Override
+ void close() throws PulsarClientException;
+
+ /**
+ * Asynchronous counterpart of {@link #close()}.
+ *
+ * @return a {@link CompletableFuture} that completes when the client has finished closing,
+ * or completes exceptionally with {@link PulsarClientException} on failure
+ */
+ CompletableFuture<Void> closeAsync();
+
+ /**
+ * Shutdown the client instance.
+ *
+ * <p>Release all resources used by the client, without waiting for pending operations to complete.
+ */
+ void shutdown();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/PulsarClientBuilder.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/PulsarClientBuilder.java
new file mode 100644
index 00000000000..101355e1182
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/PulsarClientBuilder.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import io.opentelemetry.api.OpenTelemetry;
+import java.time.Duration;
+import org.apache.pulsar.client.api.v5.auth.Authentication;
+import org.apache.pulsar.client.api.v5.config.ConnectionPolicy;
+import org.apache.pulsar.client.api.v5.config.MemorySize;
+import org.apache.pulsar.client.api.v5.config.TlsPolicy;
+import org.apache.pulsar.client.api.v5.config.TransactionPolicy;
+
+/**
+ * Builder for configuring and creating a {@link PulsarClient}.
+ */
+public interface PulsarClientBuilder {
+
+ /**
+ * Build and return the configured client.
+ *
+ * @return the configured {@link PulsarClient} instance
+ * @throws PulsarClientException if the client cannot be created (e.g., invalid configuration
+ * or connection failure)
+ */
+ PulsarClient build() throws PulsarClientException;
+
+ /**
+ * Set the Pulsar service URL (e.g., {@code pulsar://localhost:6650}).
+ *
+ * @param serviceUrl the Pulsar service URL to connect to
+ * @return this builder instance for chaining
+ */
+ PulsarClientBuilder serviceUrl(String serviceUrl);
+
+ /**
+ * Set the authentication provider.
+ *
+ * @param authentication the authentication provider to use for connecting to the broker
+ * @return this builder instance for chaining
+ */
+ PulsarClientBuilder authentication(Authentication authentication);
+
+ /**
+ * Set authentication by plugin class name and parameter string.
+ *
+ * @param authPluginClassName the fully qualified class name of the authentication plugin
+ * @param authParamsString the authentication parameters as a serialized string
+ * @return this builder instance for chaining
+ * @throws PulsarClientException if the authentication plugin cannot be loaded or configured
+ */
+ PulsarClientBuilder authentication(String authPluginClassName, String authParamsString)
+ throws PulsarClientException;
+
+ /**
+ * Timeout for client operations (e.g., creating producers/consumers).
+ *
+ * @param timeout the maximum duration to wait for an operation to complete
+ * @return this builder instance for chaining
+ */
+ PulsarClientBuilder operationTimeout(Duration timeout);
+
+ /**
+ * Configure connection-level settings such as timeouts, pool size, threading,
+ * keep-alive, and proxy configuration.
+ *
+ * @param policy the connection policy
+ * @return this builder instance for chaining
+ * @see ConnectionPolicy#builder()
+ */
+ PulsarClientBuilder connectionPolicy(ConnectionPolicy policy);
+
+ /**
+ * Set the transaction policy.
+ *
+ * @param policy the transaction policy controlling transaction behavior and timeouts
+ * @return this builder instance for chaining
+ */
+ PulsarClientBuilder transactionPolicy(TransactionPolicy policy);
+
+ /**
+ * Configure TLS for the client connection.
+ *
+ * @param policy the TLS policy to apply to broker connections
+ * @return this builder instance for chaining
+ * @see TlsPolicy#of(String)
+ * @see TlsPolicy#ofMutualTls(String, String, String)
+ * @see TlsPolicy#ofInsecure()
+ */
+ PulsarClientBuilder tlsPolicy(TlsPolicy policy);
+
+ /**
+ * Provide a custom {@link OpenTelemetry} instance for metrics and tracing.
+ *
+ * <p>If not set, the client creates its own internal instance that exports metrics
+ * (via a Prometheus-compatible endpoint) with tracing disabled.
+ *
+ * <p>When a custom instance is provided, the client uses whatever {@code MeterProvider}
+ * and {@code TracerProvider} it contains. This means:
+ * <ul>
+ * <li>To keep metrics only (no tracing), configure the instance with a
+ * {@code MeterProvider} and leave the {@code TracerProvider} as no-op.</li>
+ * <li>To enable distributed tracing, configure the instance with both a
+ * {@code MeterProvider} and a {@code TracerProvider}.</li>
+ * <li>To disable all telemetry, pass {@link OpenTelemetry#noop()}.</li>
+ * </ul>
+ *
+ * @param openTelemetry the OpenTelemetry instance to use
+ * @return this builder instance for chaining
+ */
+ PulsarClientBuilder openTelemetry(OpenTelemetry openTelemetry);
+
+ /**
+ * Maximum amount of direct memory the client can use for pending messages.
+ *
+ * @param size the memory limit for pending messages across all producers
+ * @return this builder instance for chaining
+ * @see MemorySize#ofMegabytes(long)
+ * @see MemorySize#ofGigabytes(long)
+ */
+ PulsarClientBuilder memoryLimit(MemorySize size);
+
+ // --- Misc ---
+
+ /**
+ * Set the listener name for multi-listener brokers.
+ *
+ * @param name the listener name to use when connecting to brokers that advertise
+ * multiple listener endpoints
+ * @return this builder instance for chaining
+ */
+ PulsarClientBuilder listenerName(String name);
+
+ /**
+ * A human-readable description of this client (for logging and debugging).
+ *
+ * @param description a descriptive label for this client instance
+ * @return this builder instance for chaining
+ */
+ PulsarClientBuilder description(String description);
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/PulsarClientException.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/PulsarClientException.java
new file mode 100644
index 00000000000..0e6005cec3a
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/PulsarClientException.java
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.io.IOException;
+
+/**
+ * Base exception for all Pulsar client operations.
+ */
+public class PulsarClientException extends IOException {
+
+ public PulsarClientException(String message) {
+ super(message);
+ }
+
+ public PulsarClientException(Throwable cause) {
+ super(cause);
+ }
+
+ public PulsarClientException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public static final class InvalidServiceURLException extends PulsarClientException {
+ public InvalidServiceURLException(String message) {
+ super(message);
+ }
+
+ public InvalidServiceURLException(Throwable cause) {
+ super(cause);
+ }
+ }
+
+ public static final class InvalidConfigurationException extends PulsarClientException {
+ public InvalidConfigurationException(String message) {
+ super(message);
+ }
+
+ public InvalidConfigurationException(Throwable cause) {
+ super(cause);
+ }
+ }
+
+ public static final class NotFoundException extends PulsarClientException {
+ public NotFoundException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class TimeoutException extends PulsarClientException {
+ public TimeoutException(String message) {
+ super(message);
+ }
+
+ public TimeoutException(Throwable cause) {
+ super(cause);
+ }
+ }
+
+ public static final class AlreadyClosedException extends PulsarClientException {
+ public AlreadyClosedException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class AuthenticationException extends PulsarClientException {
+ public AuthenticationException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class AuthorizationException extends PulsarClientException {
+ public AuthorizationException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class ConnectException extends PulsarClientException {
+ public ConnectException(String message) {
+ super(message);
+ }
+
+ public ConnectException(Throwable cause) {
+ super(cause);
+ }
+ }
+
+ public static final class ProducerBusyException extends PulsarClientException {
+ public ProducerBusyException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class ConsumerBusyException extends PulsarClientException {
+ public ConsumerBusyException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class ProducerQueueIsFullException extends PulsarClientException {
+ public ProducerQueueIsFullException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class IncompatibleSchemaException extends PulsarClientException {
+ public IncompatibleSchemaException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class TopicTerminatedException extends PulsarClientException {
+ public TopicTerminatedException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class CryptoException extends PulsarClientException {
+ public CryptoException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class TransactionConflictException extends PulsarClientException {
+ public TransactionConflictException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class NotConnectedException extends PulsarClientException {
+ public NotConnectedException() {
+ super("Not connected");
+ }
+
+ public NotConnectedException(String message) {
+ super(message);
+ }
+ }
+
+ public static final class MemoryBufferIsFullException extends PulsarClientException {
+ public MemoryBufferIsFullException(String message) {
+ super(message);
+ }
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/QueueConsumer.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/QueueConsumer.java
new file mode 100644
index 00000000000..e3a613e7672
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/QueueConsumer.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.time.Duration;
+import org.apache.pulsar.client.api.v5.async.AsyncQueueConsumer;
+
+/**
+ * A consumer for queue (unordered) consumption with broker-managed position tracking.
+ *
+ * <p>Messages are distributed to available consumers for parallel processing.
+ * Acknowledgment is individual: each message must be acknowledged separately.
+ *
+ * <p>This interface provides synchronous (blocking) operations. For non-blocking
+ * usage, obtain an {@link AsyncQueueConsumer} via {@link #async()}.
+ *
+ * <p>This maps to the Shared/Key_Shared subscription model in the Pulsar v4 API.
+ *
+ * @param <T> the type of message values
+ */
+public interface QueueConsumer<T> extends AutoCloseable {
+
+ /**
+ * The topic this consumer is subscribed to.
+ *
+ * @return the fully qualified topic name
+ */
+ String topic();
+
+ /**
+ * The subscription name.
+ *
+ * @return the subscription name
+ */
+ String subscription();
+
+ /**
+ * The consumer name (system-assigned or user-specified).
+ *
+ * @return the consumer name, never {@code null}
+ */
+ String consumerName();
+
+ // --- Receive ---
+
+ /**
+ * Receive a single message, blocking indefinitely.
+ *
+ * @return the received {@link Message}
+ * @throws PulsarClientException if the consumer is closed or a connection error occurs
+ */
+ Message<T> receive() throws PulsarClientException;
+
+ /**
+ * Receive a single message, blocking up to the given timeout.
+ * Returns {@code null} if the timeout elapses without a message.
+ *
+ * @param timeout the maximum time to wait for a message
+ * @return the received {@link Message}, or {@code null} if the timeout elapses
+ * @throws PulsarClientException if the consumer is closed or a connection error occurs
+ */
+ Message<T> receive(Duration timeout) throws PulsarClientException;
+
+ /**
+ * Acknowledge a single message by its ID.
+ *
+ * @param messageId the ID of the message to acknowledge
+ */
+ void acknowledge(MessageId messageId);
+
+ /**
+ * Acknowledge within a transaction. The acknowledgment becomes effective when the
+ * transaction is committed.
+ *
+ * @param messageId the ID of the message to acknowledge
+ * @param txn the transaction to associate the acknowledgment with
+ */
+ void acknowledge(MessageId messageId, Transaction txn);
+
+ /**
+ * Signal that the message with this ID could not be processed.
+ *
+ * @param messageId the ID of the message to negatively acknowledge
+ */
+ void negativeAcknowledge(MessageId messageId);
+
+ /**
+ * Return the asynchronous view of this consumer.
+ *
+ * @return the {@link AsyncQueueConsumer} counterpart of this consumer
+ */
+ AsyncQueueConsumer<T> async();
+
+ /**
+ * Close the consumer and release all resources.
+ *
+ * @throws PulsarClientException if an error occurs while closing the consumer
+ */
+ @Override
+ void close() throws PulsarClientException;
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/QueueConsumerBuilder.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/QueueConsumerBuilder.java
new file mode 100644
index 00000000000..a3f728e127e
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/QueueConsumerBuilder.java
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.regex.Pattern;
+import org.apache.pulsar.client.api.v5.config.BackoffPolicy;
+import org.apache.pulsar.client.api.v5.config.DeadLetterPolicy;
+import org.apache.pulsar.client.api.v5.config.EncryptionPolicy;
+import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
+
+/**
+ * Builder for configuring and creating a {@link QueueConsumer}.
+ *
+ * @param <T> the type of message values the consumer will receive
+ */
+public interface QueueConsumerBuilder<T> {
+
+ /**
+ * Subscribe and create the queue consumer, blocking until ready.
+ *
+ * @return the created {@link QueueConsumer}
+ * @throws PulsarClientException if the subscription fails or a connection error occurs
+ */
+ QueueConsumer<T> subscribe() throws PulsarClientException;
+
+ /**
+ * Subscribe and create the queue consumer asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes with the created {@link QueueConsumer}
+ */
+ CompletableFuture<QueueConsumer<T>> subscribeAsync();
+
+ // --- Topic selection ---
+
+ /**
+ * The topic(s) to subscribe to.
+ *
+ * @param topicNames one or more topic names
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> topic(String... topicNames);
+
+ /**
+ * The topics to subscribe to.
+ *
+ * @param topicNames the list of topic names
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> topics(List<String> topicNames);
+
+ /**
+ * Subscribe to all topics matching a regex pattern.
+ *
+ * @param pattern the compiled regex pattern to match topic names against
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> topicsPattern(Pattern pattern);
+
+ /**
+ * Subscribe to all topics matching a regex pattern (string form).
+ *
+ * @param regex the regex pattern string to match topic names against
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> topicsPattern(String regex);
+
+ // --- Subscription ---
+
+ /**
+ * The subscription name. Required for managed consumers.
+ *
+ * @param subscriptionName the subscription name
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> subscriptionName(String subscriptionName);
+
+ /**
+ * Properties to attach to the subscription.
+ *
+ * @param properties the subscription properties
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> subscriptionProperties(Map<String, String> properties);
+
+ /**
+ * Initial position when the subscription is first created.
+ *
+ * @param position the initial position
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> subscriptionInitialPosition(SubscriptionInitialPosition position);
+
+ // --- Consumer identity ---
+
+ /**
+ * A custom name for this consumer instance.
+ *
+ * @param consumerName the consumer name
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> consumerName(String consumerName);
+
+ /**
+ * Size of the receiver queue. Controls prefetch depth.
+ *
+ * @param receiverQueueSize the receiver queue size
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> receiverQueueSize(int receiverQueueSize);
+
+ /**
+ * Priority level for this consumer (lower values mean higher priority for
+ * message dispatch).
+ *
+ * @param priorityLevel the priority level
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> priorityLevel(int priorityLevel);
+
+ // --- Acknowledgment ---
+
+ /**
+ * If a message is not acknowledged within this duration, it is automatically redelivered.
+ * Set to zero to disable.
+ *
+ * @param timeout the ack timeout duration
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> ackTimeout(Duration timeout);
+
+ /**
+ * How frequently acknowledgments are flushed to the broker.
+ *
+ * @param delay the acknowledgment group time
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> acknowledgmentGroupTime(Duration delay);
+
+ /**
+ * Maximum number of acknowledgments to group before flushing.
+ *
+ * @param size the maximum acknowledgment group size
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> maxAcknowledgmentGroupSize(int size);
+
+ // --- Redelivery ---
+
+ /**
+ * Backoff strategy for redelivery after negative acknowledgment.
+ *
+ * @param backoff the backoff policy to use for negative ack redelivery
+ * @return this builder instance for chaining
+ * @see BackoffPolicy#fixed(Duration)
+ * @see BackoffPolicy#exponential(Duration, Duration)
+ */
+ QueueConsumerBuilder<T> negativeAckRedeliveryBackoff(BackoffPolicy backoff);
+
+ /**
+ * Backoff strategy for redelivery after ack timeout.
+ *
+ * @param backoff the backoff policy to use for ack timeout redelivery
+ * @return this builder instance for chaining
+ * @see BackoffPolicy#fixed(Duration)
+ * @see BackoffPolicy#exponential(Duration, Duration)
+ */
+ QueueConsumerBuilder<T> ackTimeoutRedeliveryBackoff(BackoffPolicy backoff);
+
+ // --- Dead letter queue ---
+
+ /**
+ * Configure the dead letter queue policy.
+ *
+ * @param policy the dead letter policy
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> deadLetterPolicy(DeadLetterPolicy policy);
+
+ // --- Pattern subscription ---
+
+ /**
+ * How often to re-discover topics matching the pattern.
+ *
+ * @param interval the auto-discovery interval
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> patternAutoDiscoveryPeriod(Duration interval);
+
+ // --- Encryption ---
+
+ /**
+ * Configure end-to-end message encryption for decryption.
+ *
+ * @param policy the encryption policy to use
+ * @return this builder instance for chaining
+ * @see EncryptionPolicy#forConsumer
+ */
+ QueueConsumerBuilder<T> encryptionPolicy(EncryptionPolicy policy);
+
+
+ // --- Misc ---
+
+ /**
+ * Add a single property to the consumer metadata.
+ *
+ * @param key the property key
+ * @param value the property value
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> property(String key, String value);
+
+ /**
+ * Add multiple properties to the consumer metadata.
+ *
+ * @param properties the properties to add
+ * @return this builder instance for chaining
+ */
+ QueueConsumerBuilder<T> properties(Map<String, String> properties);
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/StreamConsumer.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/StreamConsumer.java
new file mode 100644
index 00000000000..b3c29bc5c0d
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/StreamConsumer.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.io.Closeable;
+import java.time.Duration;
+import org.apache.pulsar.client.api.v5.async.AsyncStreamConsumer;
+
+/**
+ * A consumer for streaming (ordered) consumption with broker-managed position tracking.
+ *
+ * <p>Messages are delivered in order (per-key if keyed). Acknowledgment is cumulative only:
+ * acknowledging a message ID means all messages up to and including that ID are acknowledged.
+ *
+ * <p>This interface provides synchronous (blocking) operations. For non-blocking
+ * usage, obtain an {@link AsyncStreamConsumer} via {@link #async()}.
+ *
+ * <p>This maps to the Exclusive/Failover subscription model in the Pulsar v4 API.
+ *
+ * @param <T> the type of message values
+ */
+public interface StreamConsumer<T> extends Closeable {
+
+ /**
+ * The topic this consumer is subscribed to.
+ *
+ * @return the fully qualified topic name
+ */
+ String topic();
+
+ /**
+ * The subscription name.
+ *
+ * @return the subscription name
+ */
+ String subscription();
+
+ /**
+ * The consumer name (system-assigned or user-specified).
+ *
+ * @return the consumer name, never {@code null}
+ */
+ String consumerName();
+
+ /**
+ * Receive a single message, blocking indefinitely.
+ *
+ * @return the received {@link Message}
+ * @throws PulsarClientException if the consumer is closed or a connection error occurs
+ */
+ Message<T> receive() throws PulsarClientException;
+
+ /**
+ * Receive a single message, blocking up to the given timeout.
+ * Returns {@code null} if the timeout elapses without a message.
+ *
+ * @param timeout the maximum time to wait for a message
+ * @return the received {@link Message}, or {@code null} if the timeout elapses
+ * @throws PulsarClientException if the consumer is closed or a connection error occurs
+ */
+ Message<T> receive(Duration timeout) throws PulsarClientException;
+
+ /**
+ * Receive a batch of messages, blocking up to the given timeout.
+ *
+ * @param maxNumMessages the maximum number of messages to return
+ * @param timeout the maximum time to wait for messages
+ * @return the received {@link Messages} batch
+ * @throws PulsarClientException if the consumer is closed or a connection error occurs
+ */
+ Messages<T> receiveMulti(int maxNumMessages, Duration timeout) throws PulsarClientException;
+
+ /**
+ * Acknowledge all messages up to and including the given message ID.
+ *
+ * @param messageId the ID of the message to acknowledge cumulatively
+ */
+ void acknowledgeCumulative(MessageId messageId);
+
+ /**
+ * Acknowledge within a transaction. The acknowledgment becomes effective when the
+ * transaction is committed.
+ *
+ * @param messageId the ID of the message to acknowledge cumulatively
+ * @param txn the transaction to associate the acknowledgment with
+ */
+ void acknowledgeCumulative(MessageId messageId, Transaction txn);
+
+ /**
+ * Return the asynchronous view of this consumer.
+ *
+ * @return the {@link AsyncStreamConsumer} counterpart of this consumer
+ */
+ AsyncStreamConsumer<T> async();
+
+ // --- Lifecycle ---
+
+ /**
+ * Close the consumer and release all resources.
+ *
+ * @throws PulsarClientException if an error occurs while closing the consumer
+ */
+ @Override
+ void close() throws PulsarClientException;
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/StreamConsumerBuilder.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/StreamConsumerBuilder.java
new file mode 100644
index 00000000000..ad6f0df0da6
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/StreamConsumerBuilder.java
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.config.EncryptionPolicy;
+import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
+
+/**
+ * Builder for configuring and creating a {@link StreamConsumer}.
+ *
+ * @param <T> the type of message values the consumer will receive
+ */
+public interface StreamConsumerBuilder<T> {
+
+ /**
+ * Subscribe and create the stream consumer, blocking until ready.
+ *
+ * @return the created {@link StreamConsumer}
+ * @throws PulsarClientException if the subscription fails or a connection error occurs
+ */
+ StreamConsumer<T> subscribe() throws PulsarClientException;
+
+ /**
+ * Subscribe and create the stream consumer asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes with the created {@link StreamConsumer}
+ */
+ CompletableFuture<StreamConsumer<T>> subscribeAsync();
+
+ // --- Required ---
+
+ /**
+ * The topic(s) to subscribe to.
+ *
+ * @param topicNames one or more topic names
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> topic(String... topicNames);
+
+ /**
+ * The subscription name.
+ *
+ * @param subscriptionName the subscription name
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> subscriptionName(String subscriptionName);
+
+ // --- Seek (initial position override) ---
+
+ /**
+ * Reset the subscription to a specific message ID.
+ *
+ * @param messageId the message ID to seek to
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> seek(MessageId messageId);
+
+ /**
+ * Reset the subscription to a specific timestamp. The subscription
+ * will be positioned at the first message published at or after this timestamp.
+ *
+ * @param timestamp the timestamp to seek to
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> seek(Instant timestamp);
+
+ // --- Optional ---
+
+ /**
+ * Properties to attach to the subscription.
+ *
+ * @param properties the subscription properties
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> subscriptionProperties(Map<String, String> properties);
+
+ /**
+ * Initial position when the subscription is first created (no existing cursor).
+ *
+ * @param position the initial position
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> subscriptionInitialPosition(SubscriptionInitialPosition position);
+
+ /**
+ * A custom name for this consumer instance.
+ *
+ * @param consumerName the consumer name
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> consumerName(String consumerName);
+
+ /**
+ * How frequently cumulative acknowledgments are flushed to the broker.
+ *
+ * @param delay the acknowledgment group time
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> acknowledgmentGroupTime(Duration delay);
+
+ /**
+ * Whether to read from the compacted topic (only latest value per key).
+ *
+ * @param readCompacted {@code true} to read from the compacted topic
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> readCompacted(boolean readCompacted);
+
+ /**
+ * Enable replication of subscription state across geo-replicated clusters.
+ *
+ * @param replicate {@code true} to replicate subscription state
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> replicateSubscriptionState(boolean replicate);
+
+ // --- Encryption ---
+
+ /**
+ * Configure end-to-end message encryption for decryption.
+ *
+ * @param policy the encryption policy to use
+ * @return this builder instance for chaining
+ * @see EncryptionPolicy#forConsumer
+ */
+ StreamConsumerBuilder<T> encryptionPolicy(EncryptionPolicy policy);
+
+ // --- Metadata ---
+
+ /**
+ * Add a single property to the consumer metadata.
+ *
+ * @param key the property key
+ * @param value the property value
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> property(String key, String value);
+
+ /**
+ * Add multiple properties to the consumer metadata.
+ *
+ * @param properties the properties to add
+ * @return this builder instance for chaining
+ */
+ StreamConsumerBuilder<T> properties(Map<String, String> properties);
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Transaction.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Transaction.java
new file mode 100644
index 00000000000..c5fd42ff42f
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/Transaction.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import org.apache.pulsar.client.api.v5.async.AsyncTransaction;
+
+/**
+ * A Pulsar transaction handle.
+ *
+ * <p>Transactions provide exactly-once semantics across multiple topics and subscriptions.
+ * Messages produced and acknowledgments made within a transaction are atomically committed
+ * or aborted.
+ */
+public interface Transaction {
+
+ enum State {
+ OPEN,
+ COMMITTING,
+ ABORTING,
+ COMMITTED,
+ ABORTED,
+ ERROR,
+ TIMED_OUT
+ }
+
+ /**
+ * Commit this transaction, making all produced messages visible and all acknowledgments durable.
+ *
+ * @throws PulsarClientException if the transaction cannot be committed (e.g., it has already
+ * been aborted, timed out, or encountered an error)
+ */
+ void commit() throws PulsarClientException;
+
+ /**
+ * Abort this transaction, discarding all produced messages and rolling back acknowledgments.
+ *
+ * @throws PulsarClientException if the transaction cannot be aborted (e.g., it has already
+ * been committed or encountered an error)
+ */
+ void abort() throws PulsarClientException;
+
+ /**
+ * Return an asynchronous view of this transaction.
+ *
+ * @return the {@link AsyncTransaction} counterpart of this transaction
+ */
+ AsyncTransaction async();
+
+ /**
+ * The current state of this transaction.
+ *
+ * @return the current {@link State} of this transaction
+ */
+ State state();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncCheckpointConsumer.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncCheckpointConsumer.java
new file mode 100644
index 00000000000..f7bb04bfaba
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncCheckpointConsumer.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.async;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.Checkpoint;
+import org.apache.pulsar.client.api.v5.CheckpointConsumer;
+import org.apache.pulsar.client.api.v5.Message;
+
+/**
+ * Asynchronous view of a {@link CheckpointConsumer}.
+ *
+ * <p>All operations return {@link CompletableFuture} and never block.
+ * Obtained via {@link CheckpointConsumer#async()}.
+ *
+ * @param <T> the type of message values
+ */
+public interface AsyncCheckpointConsumer<T> {
+
+ /**
+ * Receive a single message asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes with the next available message
+ */
+ CompletableFuture<Message<T>> receive();
+
+ /**
+ * Receive a single message, completing with {@code null} if the timeout elapses.
+ *
+ * @param timeout the maximum duration to wait for a message
+ * @return a {@link CompletableFuture} that completes with the next available message,
+ * or {@code null} if the timeout elapses
+ */
+ CompletableFuture<Message<T>> receive(Duration timeout);
+
+ /**
+ * Receive a batch of messages asynchronously.
+ *
+ * @param maxMessages maximum number of messages to return
+ * @param timeout maximum time to wait for messages
+ * @return a {@link CompletableFuture} that completes with a list of up to
+ * {@code maxMessages} messages
+ */
+ CompletableFuture<List<Message<T>>> receiveMulti(int maxMessages, Duration timeout);
+
+ /**
+ * Create a consistent checkpoint asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes with a {@link Checkpoint} representing
+ * the current position across all segments of the topic
+ */
+ CompletableFuture<Checkpoint> checkpoint();
+
+ /**
+ * Seek to a checkpoint asynchronously.
+ *
+ * @param checkpoint the checkpoint to seek to
+ * @return a {@link CompletableFuture} that completes when the consumer has been repositioned
+ * to the given checkpoint
+ */
+ CompletableFuture<Void> seek(Checkpoint checkpoint);
+
+ /**
+ * Close this consumer asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes when the consumer has been closed
+ * and all resources have been released
+ */
+ CompletableFuture<Void> close();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncMessageBuilder.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncMessageBuilder.java
new file mode 100644
index 00000000000..4e26f8b8788
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncMessageBuilder.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.async;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.MessageMetadata;
+
+/**
+ * Asynchronous message builder, obtained from {@link AsyncProducer#newMessage()}.
+ *
+ * <p>Inherits all metadata setters from {@link MessageMetadata} and adds a
+ * non-blocking {@link #send()} terminal operation.
+ *
+ * @param <T> the type of the message value
+ */
+public interface AsyncMessageBuilder<T> extends MessageMetadata<T, AsyncMessageBuilder<T>> {
+
+ /**
+ * Send the message asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes with the {@link MessageId} assigned
+ * to the published message by the broker
+ */
+ CompletableFuture<MessageId> send();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncProducer.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncProducer.java
new file mode 100644
index 00000000000..385b0025bee
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncProducer.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.async;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.Producer;
+
+/**
+ * Asynchronous view of a {@link Producer}.
+ *
+ * <p>All operations return {@link CompletableFuture} and never block.
+ * Obtained via {@link Producer#async()}.
+ *
+ * @param <T> the type of message values this producer sends
+ */
+public interface AsyncProducer<T> {
+
+ /**
+ * Create a message builder for advanced message construction.
+ * Use {@link AsyncMessageBuilder#send()} as the terminal operation.
+ *
+ * @return a new {@link AsyncMessageBuilder} for configuring and sending a message
+ */
+ AsyncMessageBuilder<T> newMessage();
+
+ /**
+ * Flush all pending messages asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes when all pending messages have been
+ * flushed to the broker
+ */
+ CompletableFuture<Void> flush();
+
+ /**
+ * Close this producer asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes when the producer has been closed
+ * and all resources have been released
+ */
+ CompletableFuture<Void> close();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncQueueConsumer.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncQueueConsumer.java
new file mode 100644
index 00000000000..926e632f731
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncQueueConsumer.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.async;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.Message;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.QueueConsumer;
+import org.apache.pulsar.client.api.v5.Transaction;
+
+/**
+ * Asynchronous view of a {@link QueueConsumer}.
+ *
+ * <p>All operations return {@link CompletableFuture} and never block.
+ * Obtained via {@link QueueConsumer#async()}.
+ *
+ * @param <T> the type of message values
+ */
+public interface AsyncQueueConsumer<T> {
+
+ /**
+ * Receive a single message asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes with the next available message
+ */
+ CompletableFuture<Message<T>> receive();
+
+ /**
+ * Acknowledge a single message by its ID.
+ *
+ * @param messageId the ID of the message to acknowledge
+ */
+ void acknowledge(MessageId messageId);
+
+ /**
+ * Acknowledge a single message.
+ *
+ * @param message the message to acknowledge
+ */
+ void acknowledge(Message<T> message);
+
+ /**
+ * Acknowledge within a transaction. The acknowledgment becomes effective when the
+ * transaction is committed.
+ *
+ * @param messageId the ID of the message to acknowledge
+ * @param txn the transaction to associate this acknowledgment with
+ */
+ void acknowledge(MessageId messageId, Transaction txn);
+
+ /**
+ * Signal that this message could not be processed. It will be redelivered later.
+ *
+ * @param messageId the ID of the message to negatively acknowledge
+ */
+ void negativeAcknowledge(MessageId messageId);
+
+ /**
+ * Signal that this message could not be processed. It will be redelivered later.
+ *
+ * @param message the message to negatively acknowledge
+ */
+ void negativeAcknowledge(Message<T> message);
+
+ /**
+ * Close this consumer asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes when the consumer has been closed
+ * and all resources have been released
+ */
+ CompletableFuture<Void> close();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncStreamConsumer.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncStreamConsumer.java
new file mode 100644
index 00000000000..494c6f83f72
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncStreamConsumer.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.async;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.Message;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.StreamConsumer;
+import org.apache.pulsar.client.api.v5.Transaction;
+
+/**
+ * Asynchronous view of a {@link StreamConsumer}.
+ *
+ * <p>All operations return {@link CompletableFuture} and never block.
+ * Obtained via {@link StreamConsumer#async()}.
+ *
+ * @param <T> the type of message values
+ */
+public interface AsyncStreamConsumer<T> {
+
+ /**
+ * Receive a single message asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes with the next available message
+ */
+ CompletableFuture<Message<T>> receive();
+
+ /**
+ * Receive a single message, completing with {@code null} if the timeout elapses
+ * without a message becoming available.
+ *
+ * @param timeout the maximum duration to wait for a message
+ * @return a {@link CompletableFuture} that completes with the next available message,
+ * or {@code null} if the timeout elapses
+ */
+ CompletableFuture<Message<T>> receive(Duration timeout);
+
+ /**
+ * Receive a batch of messages asynchronously.
+ *
+ * @param maxNumMessages maximum number of messages to return
+ * @param timeout maximum time to wait for messages
+ * @return a {@link CompletableFuture} that completes with a list of up to
+ * {@code maxNumMessages} messages
+ */
+ CompletableFuture<List<Message<T>>> receiveMulti(int maxNumMessages, Duration timeout);
+
+ /**
+ * Acknowledge all messages up to and including the given message ID.
+ *
+ * @param messageId the message ID up to which all messages are acknowledged (inclusive)
+ */
+ void acknowledgeCumulative(MessageId messageId);
+
+ /**
+ * Acknowledge within a transaction. The acknowledgment becomes effective when the
+ * transaction is committed.
+ *
+ * @param messageId the message ID up to which all messages are acknowledged (inclusive)
+ * @param txn the transaction to associate this acknowledgment with
+ */
+ void acknowledgeCumulative(MessageId messageId, Transaction txn);
+
+ /**
+ * Close this consumer asynchronously.
+ *
+ * @return a {@link CompletableFuture} that completes when the consumer has been closed
+ * and all resources have been released
+ */
+ CompletableFuture<Void> close();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncTransaction.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncTransaction.java
new file mode 100644
index 00000000000..7dfaa6f61ea
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/AsyncTransaction.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.async;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.Transaction;
+
+/**
+ * Asynchronous view of a {@link Transaction}.
+ *
+ * <p>All operations return {@link CompletableFuture} and never block.
+ * Obtained via {@link Transaction#async()}.
+ */
+public interface AsyncTransaction {
+
+ /**
+ * Commit this transaction, making all produced messages visible and all acknowledgments durable.
+ *
+ * @return a {@link CompletableFuture} that completes when the transaction has been
+ * successfully committed
+ */
+ CompletableFuture<Void> commit();
+
+ /**
+ * Abort this transaction, discarding all produced messages and rolling back acknowledgments.
+ *
+ * @return a {@link CompletableFuture} that completes when the transaction has been
+ * successfully aborted
+ */
+ CompletableFuture<Void> abort();
+
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/package-info.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/package-info.java
new file mode 100644
index 00000000000..e8953c54c65
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/async/package-info.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Asynchronous views of producers, consumers, and transactions.
+ *
+ * <p>Each async interface mirrors its synchronous counterpart in the parent package,
+ * replacing blocking return values with {@link java.util.concurrent.CompletableFuture}.
+ * Async views are obtained via the {@code async()} accessor on the synchronous type
+ * (e.g. {@link org.apache.pulsar.client.api.v5.Producer#async()}).
+ */
+package org.apache.pulsar.client.api.v5.async;
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/Authentication.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/Authentication.java
new file mode 100644
index 00000000000..8f4f0bf19d8
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/Authentication.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.auth;
+
+import java.io.Closeable;
+import org.apache.pulsar.client.api.v5.PulsarClientException;
+
+/**
+ * Pluggable authentication provider for Pulsar clients.
+ *
+ * <p>Implementations must be thread-safe.
+ */
+public interface Authentication extends Closeable {
+
+ /**
+ * The authentication method name (e.g., "token", "tls").
+ *
+ * @return the authentication method identifier string
+ */
+ String authMethodName();
+
+ /**
+ * Get the authentication data to be sent to the broker.
+ *
+ * @return the authentication data containing credentials for the broker
+ * @throws PulsarClientException if the authentication data could not be obtained
+ */
+ AuthenticationData authData() throws PulsarClientException;
+
+ /**
+ * Get the authentication data for a specific broker host.
+ *
+ * <p>The default implementation delegates to {@link #authData()}.
+ *
+ * @param brokerHostName the hostname of the broker to authenticate against
+ * @return the authentication data containing credentials for the specified broker
+ * @throws PulsarClientException if the authentication data could not be obtained
+ */
+ default AuthenticationData authData(String brokerHostName) throws PulsarClientException {
+ return authData();
+ }
+
+ /**
+ * Initialize the authentication provider. Called once when the client is created.
+ *
+ * @throws PulsarClientException if initialization fails
+ */
+ default void initialize() throws PulsarClientException {
+ }
+
+ @Override
+ default void close() {
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/AuthenticationData.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/AuthenticationData.java
new file mode 100644
index 00000000000..c8ab63410cf
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/AuthenticationData.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.auth;
+
+import java.security.PrivateKey;
+import java.security.cert.Certificate;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Provides authentication credentials for different transport mechanisms.
+ */
+public interface AuthenticationData {
+
+ // --- HTTP authentication ---
+
+ /**
+ * Check whether this authentication data provides HTTP headers.
+ *
+ * @return {@code true} if HTTP authentication headers are available, {@code false} otherwise
+ */
+ default boolean hasDataForHttp() {
+ return false;
+ }
+
+ /**
+ * Get the HTTP authentication headers to include in requests.
+ *
+ * @return a set of header name-value entries for HTTP authentication, or an empty set if none
+ */
+ default Set<Map.Entry<String, String>> getHttpHeaders() {
+ return Set.of();
+ }
+
+ // --- TLS mutual authentication ---
+
+ /**
+ * Check whether this authentication data provides TLS client certificates.
+ *
+ * @return {@code true} if TLS certificate data is available, {@code false} otherwise
+ */
+ default boolean hasDataForTls() {
+ return false;
+ }
+
+ /**
+ * Get the TLS client certificate chain for mutual authentication.
+ *
+ * @return the client certificate chain, or {@code null} if not available
+ */
+ default Certificate[] getTlsCertificates() {
+ return null;
+ }
+
+ /**
+ * Get the TLS client private key for mutual authentication.
+ *
+ * @return the client private key, or {@code null} if not available
+ */
+ default PrivateKey getTlsPrivateKey() {
+ return null;
+ }
+
+ // --- Binary protocol authentication ---
+
+ /**
+ * Check whether this authentication data provides binary protocol command data.
+ *
+ * @return {@code true} if command data is available for the Pulsar binary protocol, {@code false} otherwise
+ */
+ default boolean hasDataFromCommand() {
+ return false;
+ }
+
+ /**
+ * Get the authentication data to include in binary protocol commands.
+ *
+ * @return the command data string for binary protocol authentication, or {@code null} if not available
+ */
+ default String getCommandData() {
+ return null;
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/AuthenticationFactory.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/AuthenticationFactory.java
new file mode 100644
index 00000000000..705b2aa429e
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/AuthenticationFactory.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.auth;
+
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.pulsar.client.api.v5.PulsarClientException;
+import org.apache.pulsar.client.api.v5.internal.PulsarClientProvider;
+
+/**
+ * Factory for creating common authentication providers.
+ */
+public final class AuthenticationFactory {
+
+ private AuthenticationFactory() {
+ }
+
+ /**
+ * Create token-based authentication with a static token.
+ *
+ * @param token the JWT or other authentication token string
+ * @return an {@link Authentication} instance configured with the given token
+ */
+ public static Authentication token(String token) {
+ return PulsarClientProvider.get().authenticationToken(token);
+ }
+
+ /**
+ * Create token-based authentication with a dynamic token supplier.
+ *
+ * <p>The supplier is invoked each time the client needs to authenticate,
+ * allowing for token refresh without recreating the client.
+ *
+ * @param tokenSupplier a supplier that provides the current authentication token
+ * @return an {@link Authentication} instance that retrieves tokens from the supplier
+ */
+ public static Authentication token(Supplier<String> tokenSupplier) {
+ return PulsarClientProvider.get().authenticationToken(tokenSupplier);
+ }
+
+ /**
+ * Create TLS mutual authentication.
+ *
+ * @param certFilePath the path to the client certificate file (PEM format)
+ * @param keyFilePath the path to the client private key file (PEM format)
+ * @return an {@link Authentication} instance configured for TLS mutual authentication
+ */
+ public static Authentication tls(String certFilePath, String keyFilePath) {
+ return PulsarClientProvider.get().authenticationTls(certFilePath, keyFilePath);
+ }
+
+ /**
+ * Create an authentication provider by plugin class name and parameter string.
+ *
+ * @param authPluginClassName the fully qualified class name of the authentication plugin
+ * @param authParamsString the authentication parameters as a serialized string
+ * @return an {@link Authentication} instance created from the specified plugin
+ * @throws PulsarClientException if the plugin class cannot be loaded or instantiated
+ */
+ public static Authentication create(String authPluginClassName, String authParamsString)
+ throws PulsarClientException {
+ return PulsarClientProvider.get().createAuthentication(authPluginClassName, authParamsString);
+ }
+
+ /**
+ * Create an authentication provider by plugin class name and parameter map.
+ *
+ * @param authPluginClassName the fully qualified class name of the authentication plugin
+ * @param authParams the authentication parameters as key-value pairs
+ * @return an {@link Authentication} instance created from the specified plugin
+ * @throws PulsarClientException if the plugin class cannot be loaded or instantiated
+ */
+ public static Authentication create(String authPluginClassName, Map<String, String> authParams)
+ throws PulsarClientException {
+ return PulsarClientProvider.get().createAuthentication(authPluginClassName, authParams);
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/CryptoFailureAction.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/CryptoFailureAction.java
new file mode 100644
index 00000000000..3b48c2a46f7
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/CryptoFailureAction.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.auth;
+
+/**
+ * Action to take when a message encryption or decryption operation fails.
+ */
+public enum CryptoFailureAction {
+
+ /**
+ * Fail the operation and return an error to the caller.
+ */
+ FAIL,
+
+ /**
+ * Silently discard the message (consumer side only).
+ */
+ DISCARD,
+
+ /**
+ * Deliver the message to the consumer without decrypting (consumer side only).
+ * The message will contain encrypted payload.
+ */
+ CONSUME
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/CryptoKeyReader.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/CryptoKeyReader.java
new file mode 100644
index 00000000000..bec3eeb6054
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/CryptoKeyReader.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.auth;
+
+import java.util.Map;
+
+/**
+ * Interface for loading encryption and decryption keys for end-to-end message encryption.
+ */
+public interface CryptoKeyReader {
+
+ /**
+ * Get the public key for encrypting messages.
+ *
+ * @param keyName the name of the key
+ * @param metadata additional metadata associated with the key
+ * @return the encryption key info containing the public key data
+ */
+ EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> metadata);
+
+ /**
+ * Get the private key for decrypting messages.
+ *
+ * @param keyName the name of the key
+ * @param metadata additional metadata associated with the key
+ * @return the encryption key info containing the private key data
+ */
+ EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> metadata);
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/EncryptionKeyInfo.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/EncryptionKeyInfo.java
new file mode 100644
index 00000000000..cc429bbc8c9
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/EncryptionKeyInfo.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.auth;
+
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Holds an encryption key and associated metadata.
+ *
+ * @param key the raw key bytes
+ * @param metadata key-value metadata associated with the key
+ */
+public record EncryptionKeyInfo(byte[] key, Map<String, String> metadata) {
+ public EncryptionKeyInfo {
+ Objects.requireNonNull(key, "key must not be null");
+ if (metadata == null) {
+ metadata = Map.of();
+ }
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/package-info.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/package-info.java
new file mode 100644
index 00000000000..7d8141ad558
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/auth/package-info.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Authentication and encryption types.
+ *
+ * <p>Provides pluggable authentication via {@link org.apache.pulsar.client.api.v5.auth.Authentication}
+ * and convenience factories in {@link org.apache.pulsar.client.api.v5.auth.AuthenticationFactory},
+ * as well as end-to-end encryption support via
+ * {@link org.apache.pulsar.client.api.v5.auth.CryptoKeyReader}.
+ */
+package org.apache.pulsar.client.api.v5.auth;
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/BackoffPolicy.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/BackoffPolicy.java
new file mode 100644
index 00000000000..eb7ab9a4ecd
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/BackoffPolicy.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * Backoff configuration for broker reconnection attempts.
+ *
+ * <p>The delay for attempt {@code n} is {@code min(initialInterval * multiplier^(n-1), maxInterval)}.
+ *
+ * @param initialInterval the delay before the first reconnection attempt
+ * @param maxInterval the maximum delay between reconnection attempts
+ * @param multiplier the multiplier applied after each attempt
+ */
+public record BackoffPolicy(
+ Duration initialInterval,
+ Duration maxInterval,
+ double multiplier
+) {
+ public BackoffPolicy {
+ Objects.requireNonNull(initialInterval, "initialInterval must not be null");
+ Objects.requireNonNull(maxInterval, "maxInterval must not be null");
+ if (multiplier < 1.0) {
+ throw new IllegalArgumentException("multiplier must be >= 1.0");
+ }
+ }
+
+ /**
+ * Create a fixed backoff (no increase between retries).
+ *
+ * @param initialInterval the constant delay between reconnection attempts
+ * @param maxInterval the maximum delay between reconnection attempts
+ * @return a {@link BackoffPolicy} with a multiplier of 1.0
+ */
+ public static BackoffPolicy fixed(Duration initialInterval, Duration maxInterval) {
+ return new BackoffPolicy(initialInterval, maxInterval, 1.0);
+ }
+
+ /**
+ * Create an exponential backoff with the given bounds and a default multiplier of 2.
+ *
+ * @param initialInterval the delay before the first reconnection attempt
+ * @param maxInterval the maximum delay between reconnection attempts
+ * @return a {@link BackoffPolicy} with a multiplier of 2.0
+ */
+ public static BackoffPolicy exponential(Duration initialInterval, Duration maxInterval) {
+ return new BackoffPolicy(initialInterval, maxInterval, 2.0);
+ }
+
+ /**
+ * Create an exponential backoff with a custom multiplier.
+ *
+ * @param initialInterval the delay before the first reconnection attempt
+ * @param maxInterval the maximum delay between reconnection attempts
+ * @param multiplier the multiplier applied after each attempt, must be >= 1.0
+ * @return a {@link BackoffPolicy} with the specified parameters
+ */
+ public static BackoffPolicy exponential(Duration initialInterval, Duration maxInterval, double multiplier) {
+ return new BackoffPolicy(initialInterval, maxInterval, multiplier);
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/BatchingPolicy.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/BatchingPolicy.java
new file mode 100644
index 00000000000..13ef7cbc4bd
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/BatchingPolicy.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+import java.time.Duration;
+
+/**
+ * Configuration for producer message batching.
+ *
+ * <p>When batching is enabled, the producer groups multiple messages into a single
+ * broker request to improve throughput. A batch is flushed when any of the
+ * configured thresholds is reached.
+ *
+ * @param enabled whether batching is enabled
+ * @param maxPublishDelay maximum time to wait before flushing a batch
+ * @param maxMessages maximum number of messages in a single batch
+ * @param maxSize maximum size of a single batch
+ */
+public record BatchingPolicy(
+ boolean enabled,
+ Duration maxPublishDelay,
+ int maxMessages,
+ MemorySize maxSize
+) {
+ public BatchingPolicy {
+ if (maxPublishDelay == null) {
+ maxPublishDelay = Duration.ofMillis(1);
+ }
+ if (maxMessages < 0) {
+ throw new IllegalArgumentException("maxMessages must be >= 0");
+ }
+ if (maxSize.bytes() < 0) {
+ throw new IllegalArgumentException("maxBytes must be >= 0");
+ }
+ }
+
+ private static final BatchingPolicy DISABLED =
+ new BatchingPolicy(false, Duration.ofMillis(1), 1000, MemorySize.ofKilobytes(128));
+ private static final BatchingPolicy DEFAULT =
+ new BatchingPolicy(true, Duration.ofMillis(1), 1000, MemorySize.ofKilobytes(128));
+
+ /**
+ * Batching disabled.
+ *
+ * @return a {@link BatchingPolicy} with batching disabled
+ */
+ public static BatchingPolicy ofDisabled() {
+ return DISABLED;
+ }
+
+ /**
+ * Batching enabled with default thresholds (1ms delay, 1000 messages, 128KB).
+ *
+ * @return a {@link BatchingPolicy} with default batching thresholds
+ */
+ public static BatchingPolicy ofDefault() {
+ return DEFAULT;
+ }
+
+ /**
+ * Batching enabled with custom thresholds.
+ *
+ * @param maxPublishDelay the maximum time to wait before flushing a batch
+ * @param maxMessages the maximum number of messages in a single batch
+ * @param maxSize the maximum size of a single batch
+ * @return a {@link BatchingPolicy} with batching enabled and the specified thresholds
+ */
+ public static BatchingPolicy of(Duration maxPublishDelay, int maxMessages, MemorySize maxSize) {
+ return new BatchingPolicy(true, maxPublishDelay, maxMessages, maxSize);
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ChunkingPolicy.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ChunkingPolicy.java
new file mode 100644
index 00000000000..ff32a287a36
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ChunkingPolicy.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+/**
+ * Configuration for chunking large messages that exceed the broker's max message size.
+ *
+ * @param enabled whether chunking is enabled
+ * @param chunkSize maximum size of each chunk in bytes
+ */
+public record ChunkingPolicy(
+ boolean enabled,
+ int chunkSize
+) {
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/CompressionPolicy.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/CompressionPolicy.java
new file mode 100644
index 00000000000..79c318211b3
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/CompressionPolicy.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+/**
+ * Compression configuration for producer message payloads.
+ *
+ * @param type the compression codec to use
+ */
+public record CompressionPolicy(CompressionType type) {
+
+ /**
+ * No compression.
+ *
+ * @return a {@link CompressionPolicy} with compression disabled
+ */
+ public static CompressionPolicy disabled() {
+ return new CompressionPolicy(CompressionType.NONE);
+ }
+
+ /**
+ * Create a compression policy with the given codec.
+ *
+ * @param type the compression codec to use for message payloads
+ * @return a {@link CompressionPolicy} configured with the specified codec
+ */
+ public static CompressionPolicy of(CompressionType type) {
+ return new CompressionPolicy(type);
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/CompressionType.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/CompressionType.java
new file mode 100644
index 00000000000..80716016a31
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/CompressionType.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+/**
+ * Compression codec used for message payloads.
+ */
+public enum CompressionType {
+
+ /** No compression. Lowest CPU cost but highest bandwidth and storage usage. */
+ NONE,
+
+ /**
+ * LZ4 compression. Very fast compression and decompression with moderate compression ratio.
+ * A good general-purpose default for latency-sensitive and CPU-sensitive workloads.
+ */
+ LZ4,
+
+ /**
+ * ZLIB compression. Higher compression ratio than LZ4 but significantly slower.
+ * Prefer {@link #ZSTD} which achieves better compression ratios with faster performance.
+ */
+ ZLIB,
+
+ /**
+ * Zstandard compression. Best compression ratio of all options with decompression speed
+ * close to LZ4. Recommended for throughput and storage-optimized workloads.
+ */
+ ZSTD,
+
+ /**
+ * Snappy compression. Very fast with a compression profile similar to LZ4.
+ * Useful for compatibility with ecosystems that standardize on Snappy.
+ */
+ SNAPPY
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ConnectionPolicy.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ConnectionPolicy.java
new file mode 100644
index 00000000000..b502dde7315
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ConnectionPolicy.java
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * Connection-level settings for the Pulsar client.
+ *
+ * <p>Groups TCP connection timeout, connection pool sizing, keep-alive, idle timeout,
+ * TCP no-delay, I/O and callback threading, and proxy configuration.
+ */
+public record ConnectionPolicy(
+ Duration connectionTimeout,
+ int connectionsPerBroker,
+ boolean enableTcpNoDelay,
+ Duration keepAliveInterval,
+ Duration connectionMaxIdleTime,
+ int ioThreads,
+ int callbackThreads,
+ String proxyServiceUrl,
+ ProxyProtocol proxyProtocol,
+ BackoffPolicy connectionBackoff
+) {
+
+ /**
+ * Create a connection policy with the given parameters.
+ *
+ * @throws NullPointerException if {@code connectionTimeout}, {@code keepAliveInterval},
+ * {@code connectionMaxIdleTime}, or {@code connectionBackoff} is null
+ * @throws IllegalArgumentException if {@code connectionsPerBroker}, {@code ioThreads},
+ * or {@code callbackThreads} is less than 1
+ */
+ public ConnectionPolicy {
+ Objects.requireNonNull(connectionTimeout, "connectionTimeout must not be null");
+ Objects.requireNonNull(keepAliveInterval, "keepAliveInterval must not be null");
+ Objects.requireNonNull(connectionMaxIdleTime, "connectionMaxIdleTime must not be null");
+ Objects.requireNonNull(connectionBackoff, "connectionBackoff must not be null");
+ if (connectionsPerBroker < 1) {
+ throw new IllegalArgumentException("connectionsPerBroker must be >= 1");
+ }
+ if (ioThreads < 1) {
+ throw new IllegalArgumentException("ioThreads must be >= 1");
+ }
+ if (callbackThreads < 1) {
+ throw new IllegalArgumentException("callbackThreads must be >= 1");
+ }
+ }
+
+ /**
+ * Create a builder for constructing a {@link ConnectionPolicy}.
+ *
+ * @return a new builder with sensible defaults
+ */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /**
+ * Builder for {@link ConnectionPolicy}.
+ */
+ public static final class Builder {
+ private Duration connectionTimeout = Duration.ofSeconds(10);
+ private int connectionsPerBroker = 1;
+ private boolean enableTcpNoDelay = true;
+ private Duration keepAliveInterval = Duration.ofSeconds(30);
+ private Duration connectionMaxIdleTime = Duration.ofMinutes(3);
+ private int ioThreads = 1;
+ private int callbackThreads = 1;
+ private String proxyServiceUrl;
+ private ProxyProtocol proxyProtocol;
+ private BackoffPolicy connectionBackoff =
+ BackoffPolicy.exponential(Duration.ofMillis(100), Duration.ofSeconds(60));
+
+ private Builder() {
+ }
+
+ /**
+ * Timeout for establishing a TCP connection to the broker.
+ *
+ * @param connectionTimeout the maximum duration to wait for a TCP connection
+ * @return this builder
+ */
+ public Builder connectionTimeout(Duration connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ return this;
+ }
+
+ /**
+ * Maximum number of TCP connections per broker.
+ *
+ * @param connectionsPerBroker the number of connections to maintain per broker
+ * @return this builder
+ */
+ public Builder connectionsPerBroker(int connectionsPerBroker) {
+ this.connectionsPerBroker = connectionsPerBroker;
+ return this;
+ }
+
+ /**
+ * Enable TCP no-delay (disable Nagle's algorithm). Default is {@code true}.
+ *
+ * @param enableTcpNoDelay {@code true} to enable TCP no-delay
+ * @return this builder
+ */
+ public Builder enableTcpNoDelay(boolean enableTcpNoDelay) {
+ this.enableTcpNoDelay = enableTcpNoDelay;
+ return this;
+ }
+
+ /**
+ * Interval for sending keep-alive probes on idle connections.
+ *
+ * @param keepAliveInterval the duration between keep-alive probes
+ * @return this builder
+ */
+ public Builder keepAliveInterval(Duration keepAliveInterval) {
+ this.keepAliveInterval = keepAliveInterval;
+ return this;
+ }
+
+ /**
+ * Maximum idle time before a connection is closed.
+ *
+ * @param connectionMaxIdleTime the maximum idle duration
+ * @return this builder
+ */
+ public Builder connectionMaxIdleTime(Duration connectionMaxIdleTime) {
+ this.connectionMaxIdleTime = connectionMaxIdleTime;
+ return this;
+ }
+
+ /**
+ * Number of I/O threads for managing connections and reading data.
+ *
+ * @param ioThreads the number of I/O threads
+ * @return this builder
+ */
+ public Builder ioThreads(int ioThreads) {
+ this.ioThreads = ioThreads;
+ return this;
+ }
+
+ /**
+ * Number of threads for message listener callbacks.
+ *
+ * @param callbackThreads the number of callback threads
+ * @return this builder
+ */
+ public Builder callbackThreads(int callbackThreads) {
+ this.callbackThreads = callbackThreads;
+ return this;
+ }
+
+ /**
+ * Connect through a proxy.
+ *
+ * @param proxyServiceUrl the URL of the proxy service
+ * @param proxyProtocol the protocol to use when connecting through the proxy
+ * @return this builder
+ */
+ public Builder proxy(String proxyServiceUrl, ProxyProtocol proxyProtocol) {
+ this.proxyServiceUrl = proxyServiceUrl;
+ this.proxyProtocol = proxyProtocol;
+ return this;
+ }
+
+ /**
+ * Backoff strategy for broker reconnection attempts.
+ *
+ * @param connectionBackoff the backoff policy to use when reconnecting to the broker
+ * @return this builder
+ * @see BackoffPolicy#exponential(Duration, Duration)
+ */
+ public Builder connectionBackoff(BackoffPolicy connectionBackoff) {
+ this.connectionBackoff = connectionBackoff;
+ return this;
+ }
+
+ /**
+ * Build the {@link ConnectionPolicy}.
+ *
+ * @return a new {@link ConnectionPolicy} instance
+ */
+ public ConnectionPolicy build() {
+ return new ConnectionPolicy(
+ connectionTimeout,
+ connectionsPerBroker,
+ enableTcpNoDelay,
+ keepAliveInterval,
+ connectionMaxIdleTime,
+ ioThreads,
+ callbackThreads,
+ proxyServiceUrl,
+ proxyProtocol,
+ connectionBackoff
+ );
+ }
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/DeadLetterPolicy.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/DeadLetterPolicy.java
new file mode 100644
index 00000000000..cb4f34ad4a4
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/DeadLetterPolicy.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+/**
+ * Configuration for the dead letter queue mechanism.
+ *
+ * <p>When a message has been redelivered more than {@code maxRedeliverCount} times,
+ * it is moved to the dead letter topic instead of being redelivered again.
+ *
+ * @param maxRedeliverCount maximum number of redelivery attempts before sending to the dead letter topic
+ * @param retryLetterTopic custom topic for retry messages (nullable, auto-generated if null)
+ * @param deadLetterTopic custom topic for dead letter messages (nullable, auto-generated if null)
+ * @param initialSubscriptionName subscription name to create on the dead letter topic (nullable)
+ */
+public record DeadLetterPolicy(
+ int maxRedeliverCount,
+ String retryLetterTopic,
+ String deadLetterTopic,
+ String initialSubscriptionName
+) {
+ public DeadLetterPolicy {
+ if (maxRedeliverCount < 0) {
+ throw new IllegalArgumentException("maxRedeliverCount must be >= 0");
+ }
+ }
+
+ /**
+ * Create a dead letter policy with just a max redeliver count, using default topic names.
+ *
+ * @param maxRedeliverCount the maximum number of redelivery attempts before sending to the dead letter topic
+ * @return a {@link DeadLetterPolicy} with auto-generated topic names and no initial subscription
+ */
+ public static DeadLetterPolicy of(int maxRedeliverCount) {
+ return new DeadLetterPolicy(maxRedeliverCount, null, null, null);
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/EncryptionPolicy.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/EncryptionPolicy.java
new file mode 100644
index 00000000000..d443274c6e7
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/EncryptionPolicy.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+import java.util.List;
+import java.util.Objects;
+import org.apache.pulsar.client.api.v5.auth.CryptoFailureAction;
+import org.apache.pulsar.client.api.v5.auth.CryptoKeyReader;
+
+/**
+ * End-to-end encryption configuration for producers and consumers.
+ *
+ * <p>For producers, supply a {@link CryptoKeyReader} and one or more encryption key names.
+ * For consumers/readers, supply a {@link CryptoKeyReader} and a {@link CryptoFailureAction}.
+ *
+ * @param keyReader the crypto key reader for loading encryption/decryption keys
+ * @param keyNames encryption key names (producer-side; empty list for consumer/reader)
+ * @param failureAction action to take when encryption or decryption fails
+ */
+public record EncryptionPolicy(
+ CryptoKeyReader keyReader,
+ List<String> keyNames,
+ CryptoFailureAction failureAction
+) {
+ public EncryptionPolicy {
+ Objects.requireNonNull(keyReader, "keyReader must not be null");
+ if (keyNames == null) {
+ keyNames = List.of();
+ }
+ if (failureAction == null) {
+ failureAction = CryptoFailureAction.FAIL;
+ }
+ }
+
+ /**
+ * Create an encryption policy for producers.
+ *
+ * @param keyReader the crypto key reader for loading encryption keys
+ * @param keyNames one or more encryption key names to use
+ * @return an {@link EncryptionPolicy} configured for producer-side encryption
+ */
+ public static EncryptionPolicy forProducer(CryptoKeyReader keyReader, String... keyNames) {
+ return new EncryptionPolicy(keyReader, List.of(keyNames), CryptoFailureAction.FAIL);
+ }
+
+ /**
+ * Create an encryption policy for consumers/readers.
+ *
+ * @param keyReader the crypto key reader for loading decryption keys
+ * @param failureAction the action to take when decryption fails
+ * @return an {@link EncryptionPolicy} configured for consumer-side decryption
+ */
+ public static EncryptionPolicy forConsumer(CryptoKeyReader keyReader, CryptoFailureAction failureAction) {
+ return new EncryptionPolicy(keyReader, List.of(), failureAction);
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/MemorySize.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/MemorySize.java
new file mode 100644
index 00000000000..8df6cbc4c52
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/MemorySize.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+/**
+ * A type-safe representation of a memory size in bytes.
+ *
+ * <p>Use the static factory methods to create instances from common units:
+ * <pre>{@code
+ * MemorySize.ofMegabytes(64) // 64 MB
+ * MemorySize.ofGigabytes(1) // 1 GB
+ * MemorySize.ofKilobytes(512) // 512 KB
+ * }</pre>
+ *
+ * @param bytes the size in bytes
+ */
+public record MemorySize(long bytes) {
+
+ public MemorySize {
+ if (bytes < 0) {
+ throw new IllegalArgumentException("bytes must be >= 0");
+ }
+ }
+
+ private static final long KB = 1024;
+ private static final long MB = 1024 * KB;
+ private static final long GB = 1024 * MB;
+
+ /**
+ * Create a memory size from a number of bytes.
+ *
+ * @param bytes the size in bytes
+ * @return a {@link MemorySize} representing the specified number of bytes
+ */
+ public static MemorySize ofBytes(long bytes) {
+ return new MemorySize(bytes);
+ }
+
+ /**
+ * Create a memory size from a number of kilobytes.
+ *
+ * @param kb the size in kilobytes
+ * @return a {@link MemorySize} representing the specified number of kilobytes
+ */
+ public static MemorySize ofKilobytes(long kb) {
+ return new MemorySize(Math.multiplyExact(kb, KB));
+ }
+
+ /**
+ * Create a memory size from a number of megabytes.
+ *
+ * @param mb the size in megabytes
+ * @return a {@link MemorySize} representing the specified number of megabytes
+ */
+ public static MemorySize ofMegabytes(long mb) {
+ return new MemorySize(Math.multiplyExact(mb, MB));
+ }
+
+ /**
+ * Create a memory size from a number of gigabytes.
+ *
+ * @param gb the size in gigabytes
+ * @return a {@link MemorySize} representing the specified number of gigabytes
+ */
+ public static MemorySize ofGigabytes(long gb) {
+ return new MemorySize(Math.multiplyExact(gb, GB));
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * @return a human-readable string representation using the largest whole unit (GB, MB, KB, or bytes)
+ */
+ @Override
+ public String toString() {
+ if (bytes >= GB && bytes % GB == 0) {
+ return bytes / GB + " GB";
+ } else if (bytes >= MB && bytes % MB == 0) {
+ return bytes / MB + " MB";
+ } else if (bytes >= KB && bytes % KB == 0) {
+ return bytes / KB + " KB";
+ }
+ return bytes + " bytes";
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ProducerAccessMode.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ProducerAccessMode.java
new file mode 100644
index 00000000000..c4836990a35
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ProducerAccessMode.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+/**
+ * Access mode for a producer on a topic.
+ */
+public enum ProducerAccessMode {
+
+ /**
+ * Multiple producers can publish on the topic.
+ */
+ SHARED,
+
+ /**
+ * Only one producer is allowed on the topic. If another producer tries to connect, it will fail.
+ */
+ EXCLUSIVE,
+
+ /**
+ * Only one producer is allowed on the topic. If another producer tries to connect,
+ * the existing producer is fenced and disconnected.
+ */
+ EXCLUSIVE_WITH_FENCING,
+
+ /**
+ * Only one producer is allowed on the topic. If another producer is already connected,
+ * the new producer will wait until it can acquire exclusive access.
+ */
+ WAIT_FOR_EXCLUSIVE
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ProxyProtocol.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ProxyProtocol.java
new file mode 100644
index 00000000000..a79fc01590b
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ProxyProtocol.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+/**
+ * Protocol used when connecting through a proxy.
+ */
+public enum ProxyProtocol {
+ SNI
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/SubscriptionInitialPosition.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/SubscriptionInitialPosition.java
new file mode 100644
index 00000000000..41fe100e271
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/SubscriptionInitialPosition.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+/**
+ * Initial position for a new subscription that has no existing cursor.
+ */
+public enum SubscriptionInitialPosition {
+
+ /**
+ * Start consuming from the latest message (the next message published after subscribing).
+ */
+ LATEST,
+
+ /**
+ * Start consuming from the earliest available message in the topic.
+ */
+ EARLIEST
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/TlsPolicy.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/TlsPolicy.java
new file mode 100644
index 00000000000..d47b8f71f73
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/TlsPolicy.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+/**
+ * TLS configuration for the Pulsar client connection.
+ *
+ * @param trustCertsFilePath path to the trusted CA certificate file (PEM format)
+ * @param keyFilePath path to the client private key file (PEM format), or {@code null}
+ * @param certificateFilePath path to the client certificate file (PEM format), or {@code null}
+ * @param allowInsecureConnection whether to allow connecting to brokers with untrusted certificates
+ * @param enableHostnameVerification whether to verify the broker hostname against the certificate
+ */
+public record TlsPolicy(
+ String trustCertsFilePath,
+ String keyFilePath,
+ String certificateFilePath,
+ boolean allowInsecureConnection,
+ boolean enableHostnameVerification
+) {
+ /**
+ * Create a TLS policy that trusts the given CA certificate and verifies hostnames.
+ *
+ * @param trustCertsFilePath the path to the trusted CA certificate file (PEM format)
+ * @return a {@link TlsPolicy} with hostname verification enabled and insecure connections disabled
+ */
+ public static TlsPolicy of(String trustCertsFilePath) {
+ return new TlsPolicy(trustCertsFilePath, null, null, false, true);
+ }
+
+ /**
+ * Create a TLS policy with mutual TLS (mTLS) authentication.
+ *
+ * @param trustCertsFilePath the path to the trusted CA certificate file (PEM format)
+ * @param keyFilePath the path to the client private key file (PEM format)
+ * @param certificateFilePath the path to the client certificate file (PEM format)
+ * @return a {@link TlsPolicy} configured for mutual TLS with hostname verification enabled
+ */
+ public static TlsPolicy ofMutualTls(String trustCertsFilePath, String keyFilePath, String certificateFilePath) {
+ return new TlsPolicy(trustCertsFilePath, keyFilePath, certificateFilePath, false, true);
+ }
+
+ /**
+ * Create an insecure TLS policy that accepts any certificate (for development only).
+ *
+ * @return a {@link TlsPolicy} with insecure connections allowed and hostname verification disabled
+ */
+ public static TlsPolicy ofInsecure() {
+ return new TlsPolicy(null, null, null, true, false);
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/TransactionPolicy.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/TransactionPolicy.java
new file mode 100644
index 00000000000..228bdeed020
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/TransactionPolicy.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.config;
+
+import java.time.Duration;
+
+/**
+ * Transaction configuration for the Pulsar client.
+ *
+ * @param timeout transaction timeout. If the transaction is not committed or aborted within this duration, it will
+ * be automatically aborted by the broker.
+ */
+public record TransactionPolicy(Duration timeout) {
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/package-info.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/package-info.java
new file mode 100644
index 00000000000..afc47322a1c
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/package-info.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Configuration value types and policy records.
+ *
+ * <p>Immutable records used to group related configuration parameters in builder APIs,
+ * such as {@link org.apache.pulsar.client.api.v5.config.BatchingPolicy},
+ * {@link org.apache.pulsar.client.api.v5.config.BackoffPolicy},
+ * {@link org.apache.pulsar.client.api.v5.config.TlsPolicy}, and
+ * {@link org.apache.pulsar.client.api.v5.config.DeadLetterPolicy}.
+ */
+package org.apache.pulsar.client.api.v5.config;
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/PulsarClientProvider.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/PulsarClientProvider.java
new file mode 100644
index 00000000000..faff56205d6
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/PulsarClientProvider.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.internal;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.function.Supplier;
+import org.apache.pulsar.client.api.v5.Checkpoint;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.PulsarClientBuilder;
+import org.apache.pulsar.client.api.v5.PulsarClientException;
+import org.apache.pulsar.client.api.v5.auth.Authentication;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+
+/**
+ * Service Provider Interface for the Pulsar client implementation.
+ *
+ * <p>The implementation module registers itself via {@code META-INF/services}
+ * using the standard Java {@link ServiceLoader} mechanism.
+ *
+ */
+public interface PulsarClientProvider {
+
+ // --- Client builder ---
+
+ PulsarClientBuilder newClientBuilder();
+
+ // --- MessageId ---
+
+ MessageId messageIdFromBytes(byte[] data) throws IOException;
+
+ MessageId earliestMessageId();
+
+ MessageId latestMessageId();
+
+ // --- Schemas ---
+
+ Schema<byte[]> bytesSchema();
+
+ Schema<String> stringSchema();
+
+ Schema<Boolean> booleanSchema();
+
+ Schema<Byte> byteSchema();
+
+ Schema<Short> shortSchema();
+
+ Schema<Integer> intSchema();
+
+ Schema<Long> longSchema();
+
+ Schema<Float> floatSchema();
+
+ Schema<Double> doubleSchema();
+
+ <T> Schema<T> jsonSchema(Class<T> pojo);
+
+ <T> Schema<T> avroSchema(Class<T> pojo);
+
+ <T extends com.google.protobuf.Message> Schema<T> protobufSchema(Class<T> clazz);
+
+ Schema<byte[]> autoProduceBytesSchema();
+
+ // --- Checkpoint ---
+
+ Checkpoint checkpointFromBytes(byte[] data) throws IOException;
+
+ Checkpoint earliestCheckpoint();
+
+ Checkpoint latestCheckpoint();
+
+ Checkpoint checkpointAtTimestamp(Instant timestamp);
+
+ // --- Authentication ---
+
+ Authentication authenticationToken(String token);
+
+ Authentication authenticationToken(Supplier<String> tokenSupplier);
+
+ Authentication authenticationTls(String certFilePath, String keyFilePath);
+
+ Authentication createAuthentication(String className, String params) throws PulsarClientException;
+
+ Authentication createAuthentication(String className, Map<String, String> params) throws PulsarClientException;
+
+ // --- Singleton access ---
+
+ static PulsarClientProvider get() {
+ return Holder.INSTANCE;
+ }
+
+ /**
+ * Lazy initialization holder for the provider singleton.
+ */
+ final class Holder {
+ static final PulsarClientProvider INSTANCE = ServiceLoader.load(PulsarClientProvider.class)
+ .findFirst()
+ .orElseThrow(() -> new IllegalStateException(
+ "No PulsarClientProvider found on the classpath. "
+ + "Add pulsar-client to your dependencies."));
+
+ private Holder() {
+ }
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/package-info.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/package-info.java
new file mode 100644
index 00000000000..da6dc6a8b0b
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/internal/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Pulsar Client API v5.
+ */
+package org.apache.pulsar.client.api.v5.internal;
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/package-info.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/package-info.java
new file mode 100644
index 00000000000..95b83146f85
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/package-info.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Pulsar Client API v5.
+ *
+ * <p>This is a redesigned client API that provides a cleaner, more modern interface for
+ * interacting with Apache Pulsar. Key design principles:
+ *
+ * <ul>
+ * <li><strong>No partition concept</strong> -- topics are opaque; only per-key ordering is
+ * guaranteed when a key is specified.</li>
+ * <li><strong>Streaming vs queuing split</strong> -- {@link org.apache.pulsar.client.api.v5.StreamConsumer}
+ * provides ordered consumption with cumulative acknowledgment, while
+ * {@link org.apache.pulsar.client.api.v5.QueueConsumer} provides parallel unordered
+ * consumption with individual acknowledgment.</li>
+ * <li><strong>Managed subscriptions</strong> -- consumers use broker-managed position tracking
+ * with automatic redelivery and dead-letter support.</li>
+ * <li><strong>Checkpoint consumer</strong> -- {@link org.apache.pulsar.client.api.v5.CheckpointConsumer}
+ * provides unmanaged consumption with consistent {@link org.apache.pulsar.client.api.v5.Checkpoint}
+ * support, designed for connector frameworks (Flink, Spark).</li>
+ * <li><strong>Modern Java</strong> -- uses {@link java.time.Duration}, {@link java.util.Optional},
+ * {@link java.time.Instant}, records, and method-reference-friendly naming.</li>
+ * <li><strong>Transactions</strong> are first-class citizens, integrated directly into the
+ * producer and consumer APIs.</li>
+ * </ul>
+ *
+ * <p>Entry point: {@link org.apache.pulsar.client.api.v5.PulsarClient#builder()}.
+ */
+package org.apache.pulsar.client.api.v5;
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/Schema.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/Schema.java
new file mode 100644
index 00000000000..da087a95b30
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/Schema.java
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.schema;
+
+import java.nio.ByteBuffer;
+import org.apache.pulsar.client.api.v5.internal.PulsarClientProvider;
+
+/**
+ * Defines how message values are serialized to bytes and deserialized from bytes.
+ *
+ * @param <T> the type of the message value
+ */
+public interface Schema<T> {
+
+ /**
+ * Encode a value to bytes.
+ *
+ * @param message the value to encode
+ * @return the serialized byte array representation of the value
+ */
+ byte[] encode(T message);
+
+ /**
+ * Decode bytes to a value.
+ *
+ * @param bytes the byte array to decode
+ * @return the deserialized value
+ */
+ T decode(byte[] bytes);
+
+ /**
+ * Decode bytes with a specific schema version. Useful for schema evolution.
+ *
+ * @param bytes the byte array to decode
+ * @param schemaVersion the schema version to use for decoding, or {@code null} for the latest version
+ * @return the deserialized value
+ */
+ default T decode(byte[] bytes, byte[] schemaVersion) {
+ return decode(bytes);
+ }
+
+ /**
+ * Decode from a ByteBuffer.
+ *
+ * @param data the {@link ByteBuffer} to decode, or {@code null}
+ * @return the deserialized value, or {@code null} if the input is {@code null}
+ */
+ default T decode(ByteBuffer data) {
+ if (data == null) {
+ return null;
+ }
+ byte[] bytes = new byte[data.remaining()];
+ data.get(bytes);
+ return decode(bytes);
+ }
+
+ /**
+ * The schema descriptor for broker-side negotiation.
+ *
+ * @return the {@link SchemaInfo} describing this schema's type and definition
+ */
+ SchemaInfo schemaInfo();
+
+ // --- Built-in schema factories ---
+
+ /**
+ * Get a schema for raw byte arrays (no serialization).
+ *
+ * @return a {@link Schema} for byte arrays
+ */
+ static Schema<byte[]> bytes() {
+ return PulsarClientProvider.get().bytesSchema();
+ }
+
+ /**
+ * Get a schema for UTF-8 strings.
+ *
+ * @return a {@link Schema} for {@link String} values
+ */
+ static Schema<String> string() {
+ return PulsarClientProvider.get().stringSchema();
+ }
+
+ /**
+ * Get a schema for boolean values.
+ *
+ * @return a {@link Schema} for {@link Boolean} values
+ */
+ static Schema<Boolean> bool() {
+ return PulsarClientProvider.get().booleanSchema();
+ }
+
+ /**
+ * Get a schema for 8-bit signed integers.
+ *
+ * @return a {@link Schema} for {@link Byte} values
+ */
+ static Schema<Byte> int8() {
+ return PulsarClientProvider.get().byteSchema();
+ }
+
+ /**
+ * Get a schema for 16-bit signed integers.
+ *
+ * @return a {@link Schema} for {@link Short} values
+ */
+ static Schema<Short> int16() {
+ return PulsarClientProvider.get().shortSchema();
+ }
+
+ /**
+ * Get a schema for 32-bit signed integers.
+ *
+ * @return a {@link Schema} for {@link Integer} values
+ */
+ static Schema<Integer> int32() {
+ return PulsarClientProvider.get().intSchema();
+ }
+
+ /**
+ * Get a schema for 64-bit signed integers.
+ *
+ * @return a {@link Schema} for {@link Long} values
+ */
+ static Schema<Long> int64() {
+ return PulsarClientProvider.get().longSchema();
+ }
+
+ /**
+ * Get a schema for 32-bit floating point numbers.
+ *
+ * @return a {@link Schema} for {@link Float} values
+ */
+ static Schema<Float> float32() {
+ return PulsarClientProvider.get().floatSchema();
+ }
+
+ /**
+ * Get a schema for 64-bit floating point numbers.
+ *
+ * @return a {@link Schema} for {@link Double} values
+ */
+ static Schema<Double> float64() {
+ return PulsarClientProvider.get().doubleSchema();
+ }
+
+ /**
+ * Get a JSON schema for a POJO class.
+ *
+ * @param <T> the POJO type
+ * @param pojo the class of the POJO to serialize and deserialize as JSON
+ * @return a {@link Schema} that encodes and decodes the POJO using JSON
+ */
+ static <T> Schema<T> json(Class<T> pojo) {
+ return PulsarClientProvider.get().jsonSchema(pojo);
+ }
+
+ /**
+ * Get an Avro schema for a POJO class.
+ *
+ * @param <T> the POJO type
+ * @param pojo the class of the POJO to serialize and deserialize using Avro
+ * @return a {@link Schema} that encodes and decodes the POJO using Avro
+ */
+ static <T> Schema<T> avro(Class<T> pojo) {
+ return PulsarClientProvider.get().avroSchema(pojo);
+ }
+
+ /**
+ * Get a Protobuf schema for a generated message class.
+ *
+ * @param <T> the Protobuf message type
+ * @param clazz the Protobuf generated message class
+ * @return a {@link Schema} that encodes and decodes the Protobuf message
+ */
+ static <T extends com.google.protobuf.Message> Schema<T> protobuf(Class<T> clazz) {
+ return PulsarClientProvider.get().protobufSchema(clazz);
+ }
+
+ /**
+ * Get a schema that auto-detects the topic schema and produces raw bytes accordingly.
+ *
+ * <p>This schema validates that the bytes being produced are compatible with the
+ * schema configured on the topic.
+ *
+ * @return a {@link Schema} for producing raw bytes with automatic schema validation
+ */
+ static Schema<byte[]> autoProduceBytes() {
+ return PulsarClientProvider.get().autoProduceBytesSchema();
+ }
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/SchemaInfo.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/SchemaInfo.java
new file mode 100644
index 00000000000..26b875f91b8
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/SchemaInfo.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.schema;
+
+import java.util.Map;
+
+/**
+ * Describes a schema for broker-side schema negotiation and compatibility checks.
+ */
+public interface SchemaInfo {
+
+ /**
+ * The name of the schema.
+ *
+ * @return the schema name
+ */
+ String name();
+
+ /**
+ * The type of the schema.
+ *
+ * @return the {@link SchemaType} indicating the serialization format
+ */
+ SchemaType type();
+
+ /**
+ * The raw schema definition bytes (e.g., Avro schema JSON, Protobuf descriptor).
+ *
+ * @return the schema definition as a byte array
+ */
+ byte[] schema();
+
+ /**
+ * Additional properties associated with the schema.
+ *
+ * @return an unmodifiable map of schema property key-value pairs
+ */
+ Map<String, String> properties();
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/SchemaType.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/SchemaType.java
new file mode 100644
index 00000000000..f80e9349ee7
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/SchemaType.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5.schema;
+
+/**
+ * Types of schema supported by Pulsar.
+ */
+public enum SchemaType {
+ NONE,
+ BYTES,
+ STRING,
+ BOOLEAN,
+ INT8,
+ INT16,
+ INT32,
+ INT64,
+ FLOAT,
+ DOUBLE,
+ JSON,
+ AVRO,
+ PROTOBUF,
+ PROTOBUF_NATIVE,
+ KEY_VALUE,
+ AUTO_CONSUME,
+ AUTO_PUBLISH
+}
diff --git a/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/package-info.java b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/package-info.java
new file mode 100644
index 00000000000..827fd8f438a
--- /dev/null
+++ b/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/schema/package-info.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Schema definitions for message serialization and deserialization.
+ *
+ * <p>The {@link org.apache.pulsar.client.api.v5.schema.Schema} interface provides static
+ * factories for built-in types (STRING, JSON, AVRO, PROTOBUF, etc.) and is the entry
+ * point for all schema-related operations.
+ */
+package org.apache.pulsar.client.api.v5.schema;
diff --git a/pulsar-client-api-v5/src/test/java/org/apache/pulsar/client/api/v5/Examples.java b/pulsar-client-api-v5/src/test/java/org/apache/pulsar/client/api/v5/Examples.java
new file mode 100644
index 00000000000..22de52c7cb2
--- /dev/null
+++ b/pulsar-client-api-v5/src/test/java/org/apache/pulsar/client/api/v5/Examples.java
@@ -0,0 +1,538 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.v5.async.AsyncProducer;
+import org.apache.pulsar.client.api.v5.async.AsyncQueueConsumer;
+import org.apache.pulsar.client.api.v5.async.AsyncStreamConsumer;
+import org.apache.pulsar.client.api.v5.auth.AuthenticationFactory;
+import org.apache.pulsar.client.api.v5.config.BackoffPolicy;
+import org.apache.pulsar.client.api.v5.config.BatchingPolicy;
+import org.apache.pulsar.client.api.v5.config.CompressionPolicy;
+import org.apache.pulsar.client.api.v5.config.CompressionType;
+import org.apache.pulsar.client.api.v5.config.ConnectionPolicy;
+import org.apache.pulsar.client.api.v5.config.DeadLetterPolicy;
+import org.apache.pulsar.client.api.v5.config.MemorySize;
+import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.v5.config.TlsPolicy;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+
+/**
+ * Usage examples for the Pulsar v5 client API.
+ *
+ * <p>These are compile-checked examples that demonstrate the API surface. They are not
+ * runnable without a real Pulsar cluster and SPI implementation.
+ */
+@SuppressWarnings("unused")
+public class Examples {
+
+ // ==================================================================================
+ // 1. Client creation
+ // ==================================================================================
+
+ /** Minimal client — connect to localhost with defaults. */
+ void minimalClient() throws Exception {
+ try (var client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .build()) {
+ // use client...
+ }
+ }
+
+ /** Production client — TLS, auth, tuned connection pool. */
+ void productionClient() throws Exception {
+ try (var client = PulsarClient.builder()
+ .serviceUrl("pulsar+ssl://pulsar.example.com:6651")
+ .authentication(AuthenticationFactory.token("eyJhbGci..."))
+ .tlsPolicy(TlsPolicy.of("/etc/pulsar/ca.pem"))
+ .operationTimeout(Duration.ofSeconds(30))
+ .connectionPolicy(ConnectionPolicy.builder()
+ .connectionTimeout(Duration.ofSeconds(10))
+ .connectionsPerBroker(2)
+ .connectionBackoff(BackoffPolicy.exponential(
+ Duration.ofMillis(100), Duration.ofSeconds(30)))
+ .build())
+ .memoryLimit(MemorySize.ofMegabytes(64))
+ .build()) {
+ // use client...
+ }
+ }
+
+ // ==================================================================================
+ // 2. Producing messages (synchronous)
+ // ==================================================================================
+
+ /** Simple produce — send strings to a topic. */
+ void simpleProducer(PulsarClient client) throws Exception {
+ try (var producer = client.newProducer(Schema.string())
+ .topic("my-topic")
+ .create()) {
+
+ // Simple send
+ producer.newMessage().value("Hello Pulsar!").send();
+
+ // Send with key and properties
+ producer.newMessage()
+ .key("user-123")
+ .value("order placed")
+ .property("orderId", "A-100")
+ .eventTime(Instant.now())
+ .send();
+ }
+ }
+
+ /** High-throughput producer — batching + compression. */
+ void highThroughputProducer(PulsarClient client) throws Exception {
+ try (var producer = client.newProducer(Schema.json(SensorReading.class))
+ .topic("sensor-data")
+ .compressionPolicy(CompressionPolicy.of(CompressionType.ZSTD))
+ .batchingPolicy(BatchingPolicy.of(
+ Duration.ofMillis(10), 5000, MemorySize.ofMegabytes(1)))
+ .create()) {
+
+ for (int i = 0; i < 100_000; i++) {
+ producer.newMessage()
+ .key("sensor-" + (i % 16))
+ .value(new SensorReading("sensor-" + i, 22.5 + i * 0.01))
+ .send();
+ }
+ }
+ }
+
+ // ==================================================================================
+ // 3. Producing messages (asynchronous)
+ // ==================================================================================
+
+ /** Async producer — fire-and-forget with flush. */
+ void asyncProducer(PulsarClient client) throws Exception {
+ try (var producer = client.newProducer(Schema.string())
+ .topic("events")
+ .create()) {
+
+ AsyncProducer<String> async = producer.async();
+
+ // Fire off many messages
+ CompletableFuture<?>[] futures = new CompletableFuture[1000];
+ for (int i = 0; i < 1000; i++) {
+ futures[i] = async.newMessage()
+ .key("key-" + (i % 10))
+ .value("event-" + i)
+ .send();
+ }
+
+ // Wait for all to complete
+ CompletableFuture.allOf(futures).join();
+
+ // Or flush to ensure everything is persisted
+ async.flush().join();
+ }
+ }
+
+ /** Async producer — pipeline with per-message callback. */
+ void asyncProducerWithCallbacks(PulsarClient client) throws Exception {
+ try (var producer = client.newProducer(Schema.string())
+ .topic("events")
+ .create()) {
+
+ for (int i = 0; i < 100; i++) {
+ final int idx = i;
+ producer.async().newMessage()
+ .value("event-" + i)
+ .send()
+ .thenAccept(msgId ->
+ System.out.printf("Message %d sent: %s%n", idx, msgId))
+ .exceptionally(ex -> {
+ System.err.printf("Message %d failed: %s%n", idx, ex.getMessage());
+ return null;
+ });
+ }
+ }
+ }
+
+ // ==================================================================================
+ // 4. Stream consumer (ordered, cumulative ack)
+ // ==================================================================================
+
+ /** Simple stream consumer — process messages in order. */
+ void streamConsumer(PulsarClient client) throws Exception {
+ try (var consumer = client.newStreamConsumer(Schema.string())
+ .topic("my-topic")
+ .subscriptionName("my-sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe()) {
+
+ while (true) {
+ Message<String> msg = consumer.receive(Duration.ofSeconds(5));
+ if (msg == null) {
+ continue; // timeout, no message
+ }
+
+ System.out.printf("[%s] key=%s value=%s%n",
+ msg.publishTime(), msg.key().orElse("(none)"), msg.value());
+
+ // Cumulative ack — everything up to this message
+ consumer.acknowledgeCumulative(msg.id());
+ }
+ }
+ }
+
+ /** Batch receive — process messages in chunks. */
+ void streamConsumerBatchReceive(PulsarClient client) throws Exception {
+ try (var consumer = client.newStreamConsumer(Schema.json(SensorReading.class))
+ .topic("sensor-data")
+ .subscriptionName("analytics")
+ .subscribe()) {
+
+ while (true) {
+ Messages<SensorReading> batch = consumer.receiveMulti(100, Duration.ofSeconds(1));
+
+ // Process the batch
+ for (var msg : batch) {
+ processSensorReading(msg.value());
+ }
+
+ // Ack up to the last message in the batch
+ consumer.acknowledgeCumulative(batch.lastId());
+ }
+ }
+ }
+
+ /** Async stream consumer — non-blocking receive loop. */
+ void asyncStreamConsumer(PulsarClient client) throws Exception {
+ try (var consumer = client.newStreamConsumer(Schema.string())
+ .topic("my-topic")
+ .subscriptionName("my-sub")
+ .subscribe()) {
+
+ AsyncStreamConsumer<String> async = consumer.async();
+
+ // Recursive async receive chain
+ receiveNext(async);
+
+ // Keep the application alive...
+ Thread.sleep(60_000);
+ }
+ }
+
+ private void receiveNext(AsyncStreamConsumer<String> async) {
+ async.receive().thenAccept(msg -> {
+ System.out.println("Received: " + msg.value());
+ async.acknowledgeCumulative(msg.id());
+
+ // Schedule next receive
+ receiveNext(async);
+ });
+ }
+
+ // ==================================================================================
+ // 5. Queue consumer (unordered, individual ack)
+ // ==================================================================================
+
+ /** Simple queue consumer — parallel processing with individual ack. */
+ void queueConsumer(PulsarClient client) throws Exception {
+ try (var consumer = client.newQueueConsumer(Schema.json(Order.class))
+ .topic("orders")
+ .subscriptionName("order-processor")
+ .subscribe()) {
+
+ while (true) {
+ Message<Order> msg = consumer.receive(Duration.ofSeconds(10));
+ if (msg == null) {
+ continue;
+ }
+
+ try {
+ processOrder(msg.value());
+ consumer.acknowledge(msg.id());
+ } catch (Exception e) {
+ // Trigger redelivery after backoff
+ consumer.negativeAcknowledge(msg.id());
+ }
+ }
+ }
+ }
+
+ /** Queue consumer with dead letter policy. */
+ void queueConsumerWithDLQ(PulsarClient client) throws Exception {
+ try (var consumer = client.newQueueConsumer(Schema.json(Order.class))
+ .topic("orders")
+ .subscriptionName("order-processor")
+ .ackTimeout(Duration.ofSeconds(30))
+ .negativeAckRedeliveryBackoff(
+ BackoffPolicy.exponential(Duration.ofSeconds(1), Duration.ofMinutes(5)))
+ .deadLetterPolicy(DeadLetterPolicy.of(5))
+ .subscribe()) {
+
+ while (true) {
+ Message<Order> msg = consumer.receive();
+ try {
+ processOrder(msg.value());
+ consumer.acknowledge(msg.id());
+ } catch (Exception e) {
+ consumer.negativeAcknowledge(msg.id());
+ // After 5 redeliveries → moves to dead letter topic automatically
+ }
+ }
+ }
+ }
+
+ /** Async queue consumer — high-throughput parallel processing. */
+ void asyncQueueConsumer(PulsarClient client) throws Exception {
+ try (var consumer = client.newQueueConsumer(Schema.json(Order.class))
+ .topic("orders")
+ .subscriptionName("parallel-processor")
+ .subscribe()) {
+
+ AsyncQueueConsumer<Order> async = consumer.async();
+
+ // Launch 10 concurrent receive loops
+ for (int i = 0; i < 10; i++) {
+ processQueueMessages(async);
+ }
+
+ Thread.sleep(60_000);
+ }
+ }
+
+ private void processQueueMessages(AsyncQueueConsumer<Order> async) {
+ async.receive().thenAccept(msg -> {
+ try {
+ processOrder(msg.value());
+ async.acknowledge(msg.id());
+ } catch (Exception e) {
+ async.negativeAcknowledge(msg.id());
+ }
+ processQueueMessages(async); // loop
+ });
+ }
+
+ // ==================================================================================
+ // 6. Transactions (exactly-once)
+ // ==================================================================================
+
+ /** Consume-transform-produce within a transaction. */
+ void transactionalProcessing(PulsarClient client) throws Exception {
+ try (var consumer = client.newQueueConsumer(Schema.json(Order.class))
+ .topic("raw-orders")
+ .subscriptionName("enricher")
+ .subscribe();
+ var producer = client.newProducer(Schema.json(EnrichedOrder.class))
+ .topic("enriched-orders")
+ .create()) {
+
+ while (true) {
+ Message<Order> msg = consumer.receive(Duration.ofSeconds(10));
+ if (msg == null) {
+ continue;
+ }
+
+ // Start transaction
+ Transaction txn = client.newTransaction();
+
+ try {
+ // Produce enriched order within the transaction
+ EnrichedOrder enriched = enrich(msg.value());
+ producer.newMessage()
+ .transaction(txn)
+ .value(enriched)
+ .send();
+
+ // Ack the source message within the same transaction
+ consumer.acknowledge(msg.id(), txn);
+
+ // Commit atomically — both the produce and ack
+ txn.commit();
+ } catch (Exception e) {
+ txn.abort();
+ }
+ }
+ }
+ }
+
+ // ==================================================================================
+ // 7. Delayed delivery
+ // ==================================================================================
+
+ /** Schedule messages for future delivery. */
+ void delayedDelivery(PulsarClient client) throws Exception {
+ try (var producer = client.newProducer(Schema.json(Reminder.class))
+ .topic("reminders")
+ .create()) {
+
+ // Deliver after a delay
+ producer.newMessage()
+ .value(new Reminder("Check status"))
+ .deliverAfter(Duration.ofMinutes(30))
+ .send();
+
+ // Deliver at a specific time
+ producer.newMessage()
+ .value(new Reminder("Daily report"))
+ .deliverAt(Instant.parse("2025-12-01T09:00:00Z"))
+ .send();
+ }
+ }
+
+ // ==================================================================================
+ // 8. Multi-topic queue consumer with pattern
+ // ==================================================================================
+
+ /** Subscribe to all topics matching a pattern. */
+ void patternSubscription(PulsarClient client) throws Exception {
+ try (var consumer = client.newQueueConsumer(Schema.string())
+ .topicsPattern("persistent://public/default/events-.*")
+ .subscriptionName("all-events")
+ .patternAutoDiscoveryPeriod(Duration.ofMinutes(1))
+ .subscribe()) {
+
+ while (true) {
+ Message<String> msg = consumer.receive();
+ System.out.printf("Topic: %s, Value: %s%n", msg.topic(), msg.value());
+ consumer.acknowledge(msg.id());
+ }
+ }
+ }
+
+ // ==================================================================================
+ // 9. Mixing sync and async on the same resource
+ // ==================================================================================
+
+ /** Use sync for setup, async for data plane. */
+ void mixedSyncAsync(PulsarClient client) throws Exception {
+ // Sync creation
+ try (var producer = client.newProducer(Schema.string())
+ .topic("my-topic")
+ .create()) {
+
+ // Sync send for the important first message
+ MessageId first = producer.newMessage()
+ .key("init")
+ .value("system started")
+ .send();
+ System.out.println("First message: " + first);
+
+ // Switch to async for the hot path
+ AsyncProducer<String> async = producer.async();
+ for (int i = 0; i < 10_000; i++) {
+ async.newMessage()
+ .value("data-" + i)
+ .send();
+ }
+ async.flush().join();
+ } // sync close waits for everything
+ }
+
+ // ==================================================================================
+ // 10. Checkpoint consumer (unmanaged, for connectors)
+ // ==================================================================================
+
+ /** Checkpoint consumer — read and checkpoint for external state management. */
+ void checkpointConsumer(PulsarClient client) throws Exception {
+ try (var consumer = client.newCheckpointConsumer(Schema.json(SensorReading.class))
+ .topic("sensor-data")
+ .startPosition(Checkpoint.earliest())
+ .create()) {
+
+ while (true) {
+ Messages<SensorReading> batch = consumer.receiveMulti(100, Duration.ofSeconds(1));
+ for (var msg : batch) {
+ processSensorReading(msg.value());
+ }
+
+ // Create a consistent checkpoint across all internal segments
+ Checkpoint cp = consumer.checkpoint();
+
+ // Serialize and store externally (e.g. Flink state backend)
+ byte[] serialized = cp.toByteArray();
+ saveToExternalState(serialized);
+ }
+ }
+ }
+
+ /** Restore a checkpoint consumer from a previously saved checkpoint. */
+ void checkpointConsumerRestore(PulsarClient client) throws Exception {
+ // Load checkpoint from external state (e.g. Flink restore)
+ byte[] saved = loadFromExternalState();
+ Checkpoint restored = Checkpoint.fromByteArray(saved);
+
+ try (var consumer = client.newCheckpointConsumer(Schema.json(SensorReading.class))
+ .topic("sensor-data")
+ .startPosition(restored)
+ .create()) {
+
+ // Continue processing from where we left off
+ while (true) {
+ Message<SensorReading> msg = consumer.receive(Duration.ofSeconds(5));
+ if (msg == null) {
+ continue;
+ }
+
+ processSensorReading(msg.value());
+ }
+ }
+ }
+
+ /** Time-travel seek — rewind to a specific timestamp. */
+ void checkpointConsumerSeek(PulsarClient client) throws Exception {
+ try (var consumer = client.newCheckpointConsumer(Schema.string())
+ .topic("events")
+ .startPosition(Checkpoint.latest())
+ .create()) {
+
+ // Seek back to replay from a specific time
+ consumer.seek(Checkpoint.atTimestamp(Instant.parse("2025-12-01T00:00:00Z")));
+
+ while (true) {
+ Message<String> msg = consumer.receive(Duration.ofSeconds(5));
+ if (msg == null) {
+ break;
+ }
+ System.out.printf("[%s] %s%n", msg.publishTime(), msg.value());
+ }
+ }
+ }
+
+ // ==================================================================================
+ // Helper types for the examples
+ // ==================================================================================
+
+ record SensorReading(String sensorId, double temperature) {}
+ record Order(String orderId, String customer, double amount) {}
+ record EnrichedOrder(String orderId, String customer, double amount, String region) {}
+ record Reminder(String text) {}
+
+ private void processSensorReading(SensorReading reading) {
+ }
+
+ private void processOrder(Order order) {
+ }
+
+ private void saveToExternalState(byte[] data) {
+ }
+
+ private byte[] loadFromExternalState() {
+ return new byte[0];
+ }
+ private EnrichedOrder enrich(Order order) {
+ return new EnrichedOrder(order.orderId, order.customer, order.amount, "us-east-1");
+ }
+}
diff --git a/settings.gradle.kts b/settings.gradle.kts
index 531914a1b42..c38e220692d 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -74,6 +74,7 @@ project(":bouncy-castle:bouncy-castle-bc").projectDir = file("bouncy-castle/bc")
include("bouncy-castle:bcfips")
include("pulsar-config-validation")
include("pulsar-client-api")
+include("pulsar-client-api-v5")
// Tier 1
include("pulsar-client-admin-api")
@@ -91,6 +92,7 @@ include("pulsar-metadata")
include("pulsar-opentelemetry")
include("pulsar-client-messagecrypto-bc")
+
// Tier 4
// Maven artifactId is "pulsar-client-admin-original" (directory is "pulsar-client-admin")
include("pulsar-client-admin-original")