diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/Proto.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/Proto.java
index f54dee18..7b870eba 100644
--- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/Proto.java
+++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/Proto.java
@@ -1,6 +1,7 @@
 // Generated by the protocol buffer compiler.  DO NOT EDIT!
 // source: person.proto
 
+// Protobuf Java Version: 3.25.6
 package org.springframework.pulsar.reactive.listener;
 
 public final class Proto {
@@ -16,35 +17,23 @@ public final class Proto {
 	}
 
 	public interface PersonOrBuilder extends
-			// @@protoc_insertion_point(interface_extends:proto.Person)
+			// @@protoc_insertion_point(interface_extends:Person)
 			com.google.protobuf.MessageOrBuilder {
 
 		/**
-		 * <code>optional int32 id = 1;</code>
-		 * @return Whether the id field is set.
-		 */
-		boolean hasId();
-
-		/**
-		 * <code>optional int32 id = 1;</code>
+		 * <code>int32 id = 2;</code>
 		 * @return The id.
 		 */
 		int getId();
 
 		/**
-		 * <code>optional string name = 2;</code>
-		 * @return Whether the name field is set.
-		 */
-		boolean hasName();
-
-		/**
-		 * <code>optional string name = 2;</code>
+		 * <code>string name = 1;</code>
 		 * @return The name.
 		 */
-		String getName();
+		java.lang.String getName();
 
 		/**
-		 * <code>optional string name = 2;</code>
+		 * <code>string name = 1;</code>
 		 * @return The bytes for name.
 		 */
 		com.google.protobuf.ByteString getNameBytes();
@@ -52,10 +41,10 @@ public final class Proto {
 	}
 
 	/**
-	 * Protobuf type {@code proto.Person}
+	 * Protobuf type {@code Person}
 	 */
 	public static final class Person extends com.google.protobuf.GeneratedMessageV3 implements
-			// @@protoc_insertion_point(message_implements:proto.Person)
+			// @@protoc_insertion_point(message_implements:Person)
 			PersonOrBuilder {
 
 		private static final long serialVersionUID = 0L;
@@ -69,144 +58,68 @@ public final class Proto {
 			name_ = "";
 		}
 
-		@Override
+		@java.lang.Override
 		@SuppressWarnings({ "unused" })
-		protected Object newInstance(UnusedPrivateParameter unused) {
+		protected java.lang.Object newInstance(UnusedPrivateParameter unused) {
 			return new Person();
 		}
 
-		@Override
-		public final com.google.protobuf.UnknownFieldSet getUnknownFields() {
-			return this.unknownFields;
-		}
-
-		private Person(com.google.protobuf.CodedInputStream input,
-				com.google.protobuf.ExtensionRegistryLite extensionRegistry)
-				throws com.google.protobuf.InvalidProtocolBufferException {
-			this();
-			if (extensionRegistry == null) {
-				throw new NullPointerException();
-			}
-			int mutable_bitField0_ = 0;
-			com.google.protobuf.UnknownFieldSet.Builder unknownFields = com.google.protobuf.UnknownFieldSet
-				.newBuilder();
-			try {
-				boolean done = false;
-				while (!done) {
-					int tag = input.readTag();
-					switch (tag) {
-						case 0:
-							done = true;
-							break;
-						case 8: {
-							bitField0_ |= 0x00000001;
-							id_ = input.readInt32();
-							break;
-						}
-						case 18: {
-							String s = input.readStringRequireUtf8();
-							bitField0_ |= 0x00000002;
-							name_ = s;
-							break;
-						}
-						default: {
-							if (!parseUnknownField(input, unknownFields, extensionRegistry, tag)) {
-								done = true;
-							}
-							break;
-						}
-					}
-				}
-			}
-			catch (com.google.protobuf.InvalidProtocolBufferException e) {
-				throw e.setUnfinishedMessage(this);
-			}
-			catch (com.google.protobuf.UninitializedMessageException e) {
-				throw e.asInvalidProtocolBufferException().setUnfinishedMessage(this);
-			}
-			catch (java.io.IOException e) {
-				throw new com.google.protobuf.InvalidProtocolBufferException(e).setUnfinishedMessage(this);
-			}
-			finally {
-				this.unknownFields = unknownFields.build();
-				makeExtensionsImmutable();
-			}
-		}
-
 		public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() {
-			return Proto.internal_static_proto_Person_descriptor;
+			return org.springframework.pulsar.reactive.listener.Proto.internal_static_Person_descriptor;
 		}
 
-		@Override
-		protected FieldAccessorTable internalGetFieldAccessorTable() {
-			return Proto.internal_static_proto_Person_fieldAccessorTable.ensureFieldAccessorsInitialized(Person.class,
-					Builder.class);
+		@java.lang.Override
+		protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internalGetFieldAccessorTable() {
+			return org.springframework.pulsar.reactive.listener.Proto.internal_static_Person_fieldAccessorTable
+				.ensureFieldAccessorsInitialized(org.springframework.pulsar.reactive.listener.Proto.Person.class,
+						org.springframework.pulsar.reactive.listener.Proto.Person.Builder.class);
 		}
 
-		private int bitField0_;
+		public static final int ID_FIELD_NUMBER = 2;
 
-		public static final int ID_FIELD_NUMBER = 1;
-
-		private int id_;
+		private int id_ = 0;
 
 		/**
-		 * <code>optional int32 id = 1;</code>
-		 * @return Whether the id field is set.
-		 */
-		@Override
-		public boolean hasId() {
-			return ((bitField0_ & 0x00000001) != 0);
-		}
-
-		/**
-		 * <code>optional int32 id = 1;</code>
+		 * <code>int32 id = 2;</code>
 		 * @return The id.
 		 */
-		@Override
+		@java.lang.Override
 		public int getId() {
 			return id_;
 		}
 
-		public static final int NAME_FIELD_NUMBER = 2;
+		public static final int NAME_FIELD_NUMBER = 1;
 
-		private volatile Object name_;
+		@SuppressWarnings("serial")
+		private volatile java.lang.Object name_ = "";
 
 		/**
-		 * <code>optional string name = 2;</code>
-		 * @return Whether the name field is set.
-		 */
-		@Override
-		public boolean hasName() {
-			return ((bitField0_ & 0x00000002) != 0);
-		}
-
-		/**
-		 * <code>optional string name = 2;</code>
+		 * <code>string name = 1;</code>
 		 * @return The name.
 		 */
-		@Override
-		public String getName() {
-			Object ref = name_;
-			if (ref instanceof String) {
-				return (String) ref;
+		@java.lang.Override
+		public java.lang.String getName() {
+			java.lang.Object ref = name_;
+			if (ref instanceof java.lang.String) {
+				return (java.lang.String) ref;
 			}
 			else {
 				com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString) ref;
-				String s = bs.toStringUtf8();
+				java.lang.String s = bs.toStringUtf8();
 				name_ = s;
 				return s;
 			}
 		}
 
 		/**
-		 * <code>optional string name = 2;</code>
+		 * <code>string name = 1;</code>
 		 * @return The bytes for name.
 		 */
-		@Override
+		@java.lang.Override
 		public com.google.protobuf.ByteString getNameBytes() {
-			Object ref = name_;
-			if (ref instanceof String) {
-				com.google.protobuf.ByteString b = com.google.protobuf.ByteString.copyFromUtf8((String) ref);
+			java.lang.Object ref = name_;
+			if (ref instanceof java.lang.String) {
+				com.google.protobuf.ByteString b = com.google.protobuf.ByteString.copyFromUtf8((java.lang.String) ref);
 				name_ = b;
 				return b;
 			}
@@ -217,7 +130,7 @@ public final class Proto {
 
 		private byte memoizedIsInitialized = -1;
 
-		@Override
+		@java.lang.Override
 		public final boolean isInitialized() {
 			byte isInitialized = memoizedIsInitialized;
 			if (isInitialized == 1)
@@ -229,142 +142,137 @@ public final class Proto {
 			return true;
 		}
 
-		@Override
+		@java.lang.Override
 		public void writeTo(com.google.protobuf.CodedOutputStream output) throws java.io.IOException {
-			if (((bitField0_ & 0x00000001) != 0)) {
-				output.writeInt32(1, id_);
+			if (!com.google.protobuf.GeneratedMessageV3.isStringEmpty(name_)) {
+				com.google.protobuf.GeneratedMessageV3.writeString(output, 1, name_);
 			}
-			if (((bitField0_ & 0x00000002) != 0)) {
-				com.google.protobuf.GeneratedMessageV3.writeString(output, 2, name_);
+			if (id_ != 0) {
+				output.writeInt32(2, id_);
 			}
-			unknownFields.writeTo(output);
+			getUnknownFields().writeTo(output);
 		}
 
-		@Override
+		@java.lang.Override
 		public int getSerializedSize() {
 			int size = memoizedSize;
 			if (size != -1)
 				return size;
 
 			size = 0;
-			if (((bitField0_ & 0x00000001) != 0)) {
-				size += com.google.protobuf.CodedOutputStream.computeInt32Size(1, id_);
+			if (!com.google.protobuf.GeneratedMessageV3.isStringEmpty(name_)) {
+				size += com.google.protobuf.GeneratedMessageV3.computeStringSize(1, name_);
 			}
-			if (((bitField0_ & 0x00000002) != 0)) {
-				size += com.google.protobuf.GeneratedMessageV3.computeStringSize(2, name_);
+			if (id_ != 0) {
+				size += com.google.protobuf.CodedOutputStream.computeInt32Size(2, id_);
 			}
-			size += unknownFields.getSerializedSize();
+			size += getUnknownFields().getSerializedSize();
 			memoizedSize = size;
 			return size;
 		}
 
-		@Override
-		public boolean equals(final Object obj) {
+		@java.lang.Override
+		public boolean equals(final java.lang.Object obj) {
 			if (obj == this) {
 				return true;
 			}
-			if (!(obj instanceof Person)) {
+			if (!(obj instanceof org.springframework.pulsar.reactive.listener.Proto.Person)) {
 				return super.equals(obj);
 			}
-			Person other = (Person) obj;
+			org.springframework.pulsar.reactive.listener.Proto.Person other = (org.springframework.pulsar.reactive.listener.Proto.Person) obj;
 
-			if (hasId() != other.hasId())
+			if (getId() != other.getId())
 				return false;
-			if (hasId()) {
-				if (getId() != other.getId())
-					return false;
-			}
-			if (hasName() != other.hasName())
+			if (!getName().equals(other.getName()))
 				return false;
-			if (hasName()) {
-				if (!getName().equals(other.getName()))
-					return false;
-			}
-			if (!unknownFields.equals(other.unknownFields))
+			if (!getUnknownFields().equals(other.getUnknownFields()))
 				return false;
 			return true;
 		}
 
-		@Override
+		@java.lang.Override
 		public int hashCode() {
 			if (memoizedHashCode != 0) {
 				return memoizedHashCode;
 			}
 			int hash = 41;
 			hash = (19 * hash) + getDescriptor().hashCode();
-			if (hasId()) {
-				hash = (37 * hash) + ID_FIELD_NUMBER;
-				hash = (53 * hash) + getId();
-			}
-			if (hasName()) {
-				hash = (37 * hash) + NAME_FIELD_NUMBER;
-				hash = (53 * hash) + getName().hashCode();
-			}
-			hash = (29 * hash) + unknownFields.hashCode();
+			hash = (37 * hash) + ID_FIELD_NUMBER;
+			hash = (53 * hash) + getId();
+			hash = (37 * hash) + NAME_FIELD_NUMBER;
+			hash = (53 * hash) + getName().hashCode();
+			hash = (29 * hash) + getUnknownFields().hashCode();
 			memoizedHashCode = hash;
 			return hash;
 		}
 
-		public static Person parseFrom(java.nio.ByteBuffer data)
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(java.nio.ByteBuffer data)
 				throws com.google.protobuf.InvalidProtocolBufferException {
 			return PARSER.parseFrom(data);
 		}
 
-		public static Person parseFrom(java.nio.ByteBuffer data,
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(java.nio.ByteBuffer data,
 				com.google.protobuf.ExtensionRegistryLite extensionRegistry)
 				throws com.google.protobuf.InvalidProtocolBufferException {
 			return PARSER.parseFrom(data, extensionRegistry);
 		}
 
-		public static Person parseFrom(com.google.protobuf.ByteString data)
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(
+				com.google.protobuf.ByteString data) throws com.google.protobuf.InvalidProtocolBufferException {
+			return PARSER.parseFrom(data);
+		}
+
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(
+				com.google.protobuf.ByteString data, com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+				throws com.google.protobuf.InvalidProtocolBufferException {
+			return PARSER.parseFrom(data, extensionRegistry);
+		}
+
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(byte[] data)
 				throws com.google.protobuf.InvalidProtocolBufferException {
 			return PARSER.parseFrom(data);
 		}
 
-		public static Person parseFrom(com.google.protobuf.ByteString data,
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(byte[] data,
 				com.google.protobuf.ExtensionRegistryLite extensionRegistry)
 				throws com.google.protobuf.InvalidProtocolBufferException {
 			return PARSER.parseFrom(data, extensionRegistry);
 		}
 
-		public static Person parseFrom(byte[] data) throws com.google.protobuf.InvalidProtocolBufferException {
-			return PARSER.parseFrom(data);
-		}
-
-		public static Person parseFrom(byte[] data, com.google.protobuf.ExtensionRegistryLite extensionRegistry)
-				throws com.google.protobuf.InvalidProtocolBufferException {
-			return PARSER.parseFrom(data, extensionRegistry);
-		}
-
-		public static Person parseFrom(java.io.InputStream input) throws java.io.IOException {
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(java.io.InputStream input)
+				throws java.io.IOException {
 			return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input);
 		}
 
-		public static Person parseFrom(java.io.InputStream input,
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(java.io.InputStream input,
 				com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException {
 			return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input, extensionRegistry);
 		}
 
-		public static Person parseDelimitedFrom(java.io.InputStream input) throws java.io.IOException {
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseDelimitedFrom(
+				java.io.InputStream input) throws java.io.IOException {
 			return com.google.protobuf.GeneratedMessageV3.parseDelimitedWithIOException(PARSER, input);
 		}
 
-		public static Person parseDelimitedFrom(java.io.InputStream input,
-				com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException {
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseDelimitedFrom(
+				java.io.InputStream input, com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+				throws java.io.IOException {
 			return com.google.protobuf.GeneratedMessageV3.parseDelimitedWithIOException(PARSER, input,
 					extensionRegistry);
 		}
 
-		public static Person parseFrom(com.google.protobuf.CodedInputStream input) throws java.io.IOException {
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(
+				com.google.protobuf.CodedInputStream input) throws java.io.IOException {
 			return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input);
 		}
 
-		public static Person parseFrom(com.google.protobuf.CodedInputStream input,
-				com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException {
+		public static org.springframework.pulsar.reactive.listener.Proto.Person parseFrom(
+				com.google.protobuf.CodedInputStream input, com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+				throws java.io.IOException {
 			return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input, extensionRegistry);
 		}
 
-		@Override
+		@java.lang.Override
 		public Builder newBuilderForType() {
 			return newBuilder();
 		}
@@ -373,136 +281,135 @@ public final class Proto {
 			return DEFAULT_INSTANCE.toBuilder();
 		}
 
-		public static Builder newBuilder(Person prototype) {
+		public static Builder newBuilder(org.springframework.pulsar.reactive.listener.Proto.Person prototype) {
 			return DEFAULT_INSTANCE.toBuilder().mergeFrom(prototype);
 		}
 
-		@Override
+		@java.lang.Override
 		public Builder toBuilder() {
 			return this == DEFAULT_INSTANCE ? new Builder() : new Builder().mergeFrom(this);
 		}
 
-		@Override
-		protected Builder newBuilderForType(BuilderParent parent) {
+		@java.lang.Override
+		protected Builder newBuilderForType(com.google.protobuf.GeneratedMessageV3.BuilderParent parent) {
 			Builder builder = new Builder(parent);
 			return builder;
 		}
 
 		/**
-		 * Protobuf type {@code proto.Person}
+		 * Protobuf type {@code Person}
 		 */
 		public static final class Builder extends com.google.protobuf.GeneratedMessageV3.Builder<Builder> implements
-				// @@protoc_insertion_point(builder_implements:proto.Person)
-				PersonOrBuilder {
+				// @@protoc_insertion_point(builder_implements:Person)
+				org.springframework.pulsar.reactive.listener.Proto.PersonOrBuilder {
 
 			public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() {
-				return Proto.internal_static_proto_Person_descriptor;
+				return org.springframework.pulsar.reactive.listener.Proto.internal_static_Person_descriptor;
 			}
 
-			@Override
-			protected FieldAccessorTable internalGetFieldAccessorTable() {
-				return Proto.internal_static_proto_Person_fieldAccessorTable
-					.ensureFieldAccessorsInitialized(Person.class, Builder.class);
+			@java.lang.Override
+			protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internalGetFieldAccessorTable() {
+				return org.springframework.pulsar.reactive.listener.Proto.internal_static_Person_fieldAccessorTable
+					.ensureFieldAccessorsInitialized(org.springframework.pulsar.reactive.listener.Proto.Person.class,
+							org.springframework.pulsar.reactive.listener.Proto.Person.Builder.class);
 			}
 
 			// Construct using
-			// org.springframework.pulsar.listener.Proto.Person.newBuilder()
+			// org.springframework.pulsar.reactive.listener.Proto.Person.newBuilder()
 			private Builder() {
-				maybeForceBuilderInitialization();
+
 			}
 
-			private Builder(BuilderParent parent) {
+			private Builder(com.google.protobuf.GeneratedMessageV3.BuilderParent parent) {
 				super(parent);
-				maybeForceBuilderInitialization();
+
 			}
 
-			private void maybeForceBuilderInitialization() {
-				if (com.google.protobuf.GeneratedMessageV3.alwaysUseFieldBuilders) {
-				}
-			}
-
-			@Override
+			@java.lang.Override
 			public Builder clear() {
 				super.clear();
+				bitField0_ = 0;
 				id_ = 0;
-				bitField0_ = (bitField0_ & ~0x00000001);
 				name_ = "";
-				bitField0_ = (bitField0_ & ~0x00000002);
 				return this;
 			}
 
-			@Override
+			@java.lang.Override
 			public com.google.protobuf.Descriptors.Descriptor getDescriptorForType() {
-				return Proto.internal_static_proto_Person_descriptor;
+				return org.springframework.pulsar.reactive.listener.Proto.internal_static_Person_descriptor;
 			}
 
-			@Override
-			public Person getDefaultInstanceForType() {
-				return Person.getDefaultInstance();
+			@java.lang.Override
+			public org.springframework.pulsar.reactive.listener.Proto.Person getDefaultInstanceForType() {
+				return org.springframework.pulsar.reactive.listener.Proto.Person.getDefaultInstance();
 			}
 
-			@Override
-			public Person build() {
-				Person result = buildPartial();
+			@java.lang.Override
+			public org.springframework.pulsar.reactive.listener.Proto.Person build() {
+				org.springframework.pulsar.reactive.listener.Proto.Person result = buildPartial();
 				if (!result.isInitialized()) {
 					throw newUninitializedMessageException(result);
 				}
 				return result;
 			}
 
-			@Override
-			public Person buildPartial() {
-				Person result = new Person(this);
-				int from_bitField0_ = bitField0_;
-				int to_bitField0_ = 0;
-				if (((from_bitField0_ & 0x00000001) != 0)) {
-					result.id_ = id_;
-					to_bitField0_ |= 0x00000001;
+			@java.lang.Override
+			public org.springframework.pulsar.reactive.listener.Proto.Person buildPartial() {
+				org.springframework.pulsar.reactive.listener.Proto.Person result = new org.springframework.pulsar.reactive.listener.Proto.Person(
+						this);
+				if (bitField0_ != 0) {
+					buildPartial0(result);
 				}
-				if (((from_bitField0_ & 0x00000002) != 0)) {
-					to_bitField0_ |= 0x00000002;
-				}
-				result.name_ = name_;
-				result.bitField0_ = to_bitField0_;
 				onBuilt();
 				return result;
 			}
 
-			@Override
+			private void buildPartial0(org.springframework.pulsar.reactive.listener.Proto.Person result) {
+				int from_bitField0_ = bitField0_;
+				if (((from_bitField0_ & 0x00000001) != 0)) {
+					result.id_ = id_;
+				}
+				if (((from_bitField0_ & 0x00000002) != 0)) {
+					result.name_ = name_;
+				}
+			}
+
+			@java.lang.Override
 			public Builder clone() {
 				return super.clone();
 			}
 
-			@Override
-			public Builder setField(com.google.protobuf.Descriptors.FieldDescriptor field, Object value) {
+			@java.lang.Override
+			public Builder setField(com.google.protobuf.Descriptors.FieldDescriptor field, java.lang.Object value) {
 				return super.setField(field, value);
 			}
 
-			@Override
+			@java.lang.Override
 			public Builder clearField(com.google.protobuf.Descriptors.FieldDescriptor field) {
 				return super.clearField(field);
 			}
 
-			@Override
+			@java.lang.Override
 			public Builder clearOneof(com.google.protobuf.Descriptors.OneofDescriptor oneof) {
 				return super.clearOneof(oneof);
 			}
 
-			@Override
+			@java.lang.Override
 			public Builder setRepeatedField(com.google.protobuf.Descriptors.FieldDescriptor field, int index,
-					Object value) {
+					java.lang.Object value) {
 				return super.setRepeatedField(field, index, value);
 			}
 
-			@Override
-			public Builder addRepeatedField(com.google.protobuf.Descriptors.FieldDescriptor field, Object value) {
+			@java.lang.Override
+			public Builder addRepeatedField(com.google.protobuf.Descriptors.FieldDescriptor field,
+					java.lang.Object value) {
 				return super.addRepeatedField(field, value);
 			}
 
-			@Override
+			@java.lang.Override
 			public Builder mergeFrom(com.google.protobuf.Message other) {
-				if (other instanceof Person) {
-					return mergeFrom((Person) other);
+				if (other instanceof org.springframework.pulsar.reactive.listener.Proto.Person) {
+					return mergeFrom((org.springframework.pulsar.reactive.listener.Proto.Person) other);
 				}
 				else {
 					super.mergeFrom(other);
@@ -510,43 +417,66 @@ public final class Proto {
 				}
 			}
 
-			public Builder mergeFrom(Person other) {
-				if (other == Person.getDefaultInstance())
+			public Builder mergeFrom(org.springframework.pulsar.reactive.listener.Proto.Person other) {
+				if (other == org.springframework.pulsar.reactive.listener.Proto.Person.getDefaultInstance())
 					return this;
-				if (other.hasId()) {
+				if (other.getId() != 0) {
 					setId(other.getId());
 				}
-				if (other.hasName()) {
-					bitField0_ |= 0x00000002;
+				if (!other.getName().isEmpty()) {
 					name_ = other.name_;
+					bitField0_ |= 0x00000002;
 					onChanged();
 				}
-				this.mergeUnknownFields(other.unknownFields);
+				this.mergeUnknownFields(other.getUnknownFields());
 				onChanged();
 				return this;
 			}
 
-			@Override
+			@java.lang.Override
 			public final boolean isInitialized() {
 				return true;
 			}
 
-			@Override
+			@java.lang.Override
 			public Builder mergeFrom(com.google.protobuf.CodedInputStream input,
 					com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException {
-				Person parsedMessage = null;
+				if (extensionRegistry == null) {
+					throw new java.lang.NullPointerException();
+				}
 				try {
-					parsedMessage = PARSER.parsePartialFrom(input, extensionRegistry);
+					boolean done = false;
+					while (!done) {
+						int tag = input.readTag();
+						switch (tag) {
+							case 0:
+								done = true;
+								break;
+							case 10: {
+								name_ = input.readStringRequireUtf8();
+								bitField0_ |= 0x00000002;
+								break;
+							} // case 10
+							case 16: {
+								id_ = input.readInt32();
+								bitField0_ |= 0x00000001;
+								break;
+							} // case 16
+							default: {
+								if (!super.parseUnknownField(input, extensionRegistry, tag)) {
+									done = true; // was an endgroup tag
+								}
+								break;
+							} // default:
+						} // switch (tag)
+					} // while (!done)
 				}
 				catch (com.google.protobuf.InvalidProtocolBufferException e) {
-					parsedMessage = (Person) e.getUnfinishedMessage();
 					throw e.unwrapIOException();
 				}
 				finally {
-					if (parsedMessage != null) {
-						mergeFrom(parsedMessage);
-					}
-				}
+					onChanged();
+				} // finally
 				return this;
 			}
 
@@ -555,37 +485,29 @@ public final class Proto {
 			private int id_;
 
 			/**
-			 * <code>optional int32 id = 1;</code>
-			 * @return Whether the id field is set.
-			 */
-			@Override
-			public boolean hasId() {
-				return ((bitField0_ & 0x00000001) != 0);
-			}
-
-			/**
-			 * <code>optional int32 id = 1;</code>
+			 * <code>int32 id = 2;</code>
 			 * @return The id.
 			 */
-			@Override
+			@java.lang.Override
 			public int getId() {
 				return id_;
 			}
 
 			/**
-			 * <code>optional int32 id = 1;</code>
+			 * <code>int32 id = 2;</code>
 			 * @param value The id to set.
 			 * @return This builder for chaining.
 			 */
 			public Builder setId(int value) {
-				bitField0_ |= 0x00000001;
+
 				id_ = value;
+				bitField0_ |= 0x00000001;
 				onChanged();
 				return this;
 			}
 
 			/**
-			 * <code>optional int32 id = 1;</code>
+			 * <code>int32 id = 2;</code>
 			 * @return This builder for chaining.
 			 */
 			public Builder clearId() {
@@ -595,41 +517,34 @@ public final class Proto {
 				return this;
 			}
 
-			private Object name_ = "";
+			private java.lang.Object name_ = "";
 
 			/**
-			 * <code>optional string name = 2;</code>
-			 * @return Whether the name field is set.
-			 */
-			public boolean hasName() {
-				return ((bitField0_ & 0x00000002) != 0);
-			}
-
-			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @return The name.
 			 */
-			public String getName() {
-				Object ref = name_;
-				if (!(ref instanceof String)) {
+			public java.lang.String getName() {
+				java.lang.Object ref = name_;
+				if (!(ref instanceof java.lang.String)) {
 					com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString) ref;
-					String s = bs.toStringUtf8();
+					java.lang.String s = bs.toStringUtf8();
 					name_ = s;
 					return s;
 				}
 				else {
-					return (String) ref;
+					return (java.lang.String) ref;
 				}
 			}
 
 			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @return The bytes for name.
 			 */
 			public com.google.protobuf.ByteString getNameBytes() {
-				Object ref = name_;
+				java.lang.Object ref = name_;
 				if (ref instanceof String) {
-					com.google.protobuf.ByteString b = com.google.protobuf.ByteString.copyFromUtf8((String) ref);
+					com.google.protobuf.ByteString b = com.google.protobuf.ByteString
+						.copyFromUtf8((java.lang.String) ref);
 					name_ = b;
 					return b;
 				}
@@ -639,33 +554,33 @@ public final class Proto {
 			}
 
 			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @param value The name to set.
 			 * @return This builder for chaining.
 			 */
-			public Builder setName(String value) {
+			public Builder setName(java.lang.String value) {
 				if (value == null) {
 					throw new NullPointerException();
 				}
-				bitField0_ |= 0x00000002;
 				name_ = value;
+				bitField0_ |= 0x00000002;
 				onChanged();
 				return this;
 			}
 
 			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @return This builder for chaining.
 			 */
 			public Builder clearName() {
-				bitField0_ = (bitField0_ & ~0x00000002);
 				name_ = getDefaultInstance().getName();
+				bitField0_ = (bitField0_ & ~0x00000002);
 				onChanged();
 				return this;
 			}
 
 			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @param value The bytes for name to set.
 			 * @return This builder for chaining.
 			 */
@@ -674,42 +589,56 @@ public final class Proto {
 					throw new NullPointerException();
 				}
 				checkByteStringIsUtf8(value);
-				bitField0_ |= 0x00000002;
 				name_ = value;
+				bitField0_ |= 0x00000002;
 				onChanged();
 				return this;
 			}
 
-			@Override
+			@java.lang.Override
 			public final Builder setUnknownFields(final com.google.protobuf.UnknownFieldSet unknownFields) {
 				return super.setUnknownFields(unknownFields);
 			}
 
-			@Override
+			@java.lang.Override
 			public final Builder mergeUnknownFields(final com.google.protobuf.UnknownFieldSet unknownFields) {
 				return super.mergeUnknownFields(unknownFields);
 			}
 
-			// @@protoc_insertion_point(builder_scope:proto.Person)
+			// @@protoc_insertion_point(builder_scope:Person)
 
 		}
 
-		// @@protoc_insertion_point(class_scope:proto.Person)
-		private static final Person DEFAULT_INSTANCE;
+		// @@protoc_insertion_point(class_scope:Person)
+		private static final org.springframework.pulsar.reactive.listener.Proto.Person DEFAULT_INSTANCE;
 		static {
-			DEFAULT_INSTANCE = new Person();
+			DEFAULT_INSTANCE = new org.springframework.pulsar.reactive.listener.Proto.Person();
 		}
 
-		public static Person getDefaultInstance() {
+		public static org.springframework.pulsar.reactive.listener.Proto.Person getDefaultInstance() {
 			return DEFAULT_INSTANCE;
 		}
 
 		private static final com.google.protobuf.Parser<Person> PARSER = new com.google.protobuf.AbstractParser<Person>() {
-			@Override
+			@java.lang.Override
 			public Person parsePartialFrom(com.google.protobuf.CodedInputStream input,
 					com.google.protobuf.ExtensionRegistryLite extensionRegistry)
 					throws com.google.protobuf.InvalidProtocolBufferException {
-				return new Person(input, extensionRegistry);
+				Builder builder = newBuilder();
+				try {
+					builder.mergeFrom(input, extensionRegistry);
+				}
+				catch (com.google.protobuf.InvalidProtocolBufferException e) {
+					throw e.setUnfinishedMessage(builder.buildPartial());
+				}
+				catch (com.google.protobuf.UninitializedMessageException e) {
+					throw e.asInvalidProtocolBufferException().setUnfinishedMessage(builder.buildPartial());
+				}
+				catch (java.io.IOException e) {
+					throw new com.google.protobuf.InvalidProtocolBufferException(e)
+						.setUnfinishedMessage(builder.buildPartial());
+				}
+				return builder.buildPartial();
 			}
 		};
 
@@ -717,21 +646,21 @@ public final class Proto {
 			return PARSER;
 		}
 
-		@Override
+		@java.lang.Override
 		public com.google.protobuf.Parser<Person> getParserForType() {
 			return PARSER;
 		}
 
-		@Override
-		public Person getDefaultInstanceForType() {
+		@java.lang.Override
+		public org.springframework.pulsar.reactive.listener.Proto.Person getDefaultInstanceForType() {
 			return DEFAULT_INSTANCE;
 		}
 
 	}
 
-	private static final com.google.protobuf.Descriptors.Descriptor internal_static_proto_Person_descriptor;
+	private static final com.google.protobuf.Descriptors.Descriptor internal_static_Person_descriptor;
 
-	private static final com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internal_static_proto_Person_fieldAccessorTable;
+	private static final com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internal_static_Person_fieldAccessorTable;
 
 	public static com.google.protobuf.Descriptors.FileDescriptor getDescriptor() {
 		return descriptor;
@@ -739,14 +668,15 @@ public final class Proto {
 
 	private static com.google.protobuf.Descriptors.FileDescriptor descriptor;
 	static {
-		String[] descriptorData = { "\n\014person.proto\022\005proto\"<\n\006Person\022\017\n\002id\030\001 "
-				+ "\001(\005H\000\210\001\001\022\021\n\004name\030\002 \001(\tH\001\210\001\001B\005\n\003_idB\007\n\005_n"
-				+ "ameB,\n#org.springframework.pulsar.listen" + "erB\005Protob\006proto3" };
+		java.lang.String[] descriptorData = {
+				"\n\014person.proto\"\"\n\006Person\022\n\n\002id\030\002 \001(\005\022\014\n\004"
+						+ "name\030\001 \001(\tB5\n,org.springframework.pulsar"
+						+ ".reactive.listenerB\005Protob\006proto3" };
 		descriptor = com.google.protobuf.Descriptors.FileDescriptor.internalBuildGeneratedFileFrom(descriptorData,
 				new com.google.protobuf.Descriptors.FileDescriptor[] {});
-		internal_static_proto_Person_descriptor = getDescriptor().getMessageTypes().get(0);
-		internal_static_proto_Person_fieldAccessorTable = new com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
-				internal_static_proto_Person_descriptor, new String[] { "Id", "Name", "Id", "Name", });
+		internal_static_Person_descriptor = getDescriptor().getMessageTypes().get(0);
+		internal_static_Person_fieldAccessorTable = new com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
+				internal_static_Person_descriptor, new java.lang.String[] { "Id", "Name", });
 	}
 
 	// @@protoc_insertion_point(outer_class_scope)
diff --git a/spring-pulsar-reactive/src/test/proto/README.adoc b/spring-pulsar-reactive/src/test/proto/README.adoc
new file mode 100644
index 00000000..2c693b3d
--- /dev/null
+++ b/spring-pulsar-reactive/src/test/proto/README.adoc
@@ -0,0 +1 @@
+include::../../../../spring-pulsar/src/test/proto/README.adoc[]
diff --git a/spring-pulsar-reactive/src/test/proto/person.proto b/spring-pulsar-reactive/src/test/proto/person.proto
new file mode 100644
index 00000000..c624c759
--- /dev/null
+++ b/spring-pulsar-reactive/src/test/proto/person.proto
@@ -0,0 +1,9 @@
+syntax = "proto3";
+
+option java_package = "org.springframework.pulsar.reactive.listener";
+option java_outer_classname = "Proto";
+
+message Person {
+  int32 id = 2;
+  string name = 1;
+}
diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/Proto.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/Proto.java
index 10f4c052..c1915a53 100644
--- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/Proto.java
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/Proto.java
@@ -1,6 +1,7 @@
 // Generated by the protocol buffer compiler.  DO NOT EDIT!
 // source: person.proto
 
+// Protobuf Java Version: 3.25.6
 package org.springframework.pulsar.listener;
 
 public final class Proto {
@@ -16,35 +17,23 @@ public final class Proto {
 	}
 
 	public interface PersonOrBuilder extends
-			// @@protoc_insertion_point(interface_extends:proto.Person)
+			// @@protoc_insertion_point(interface_extends:Person)
 			com.google.protobuf.MessageOrBuilder {
 
 		/**
-		 * <code>optional int32 id = 1;</code>
-		 * @return Whether the id field is set.
-		 */
-		boolean hasId();
-
-		/**
-		 * <code>optional int32 id = 1;</code>
+		 * <code>int32 id = 2;</code>
 		 * @return The id.
 		 */
 		int getId();
 
 		/**
-		 * <code>optional string name = 2;</code>
-		 * @return Whether the name field is set.
-		 */
-		boolean hasName();
-
-		/**
-		 * <code>optional string name = 2;</code>
+		 * <code>string name = 1;</code>
 		 * @return The name.
 		 */
 		java.lang.String getName();
 
 		/**
-		 * <code>optional string name = 2;</code>
+		 * <code>string name = 1;</code>
 		 * @return The bytes for name.
 		 */
 		com.google.protobuf.ByteString getNameBytes();
@@ -52,10 +41,10 @@ public final class Proto {
 	}
 
 	/**
-	 * Protobuf type {@code proto.Person}
+	 * Protobuf type {@code Person}
 	 */
 	public static final class Person extends com.google.protobuf.GeneratedMessageV3 implements
-			// @@protoc_insertion_point(message_implements:proto.Person)
+			// @@protoc_insertion_point(message_implements:Person)
 			PersonOrBuilder {
 
 		private static final long serialVersionUID = 0L;
@@ -75,92 +64,23 @@ public final class Proto {
 			return new Person();
 		}
 
-		@java.lang.Override
-		public final com.google.protobuf.UnknownFieldSet getUnknownFields() {
-			return this.unknownFields;
-		}
-
-		private Person(com.google.protobuf.CodedInputStream input,
-				com.google.protobuf.ExtensionRegistryLite extensionRegistry)
-				throws com.google.protobuf.InvalidProtocolBufferException {
-			this();
-			if (extensionRegistry == null) {
-				throw new java.lang.NullPointerException();
-			}
-			int mutable_bitField0_ = 0;
-			com.google.protobuf.UnknownFieldSet.Builder unknownFields = com.google.protobuf.UnknownFieldSet
-				.newBuilder();
-			try {
-				boolean done = false;
-				while (!done) {
-					int tag = input.readTag();
-					switch (tag) {
-						case 0:
-							done = true;
-							break;
-						case 8: {
-							bitField0_ |= 0x00000001;
-							id_ = input.readInt32();
-							break;
-						}
-						case 18: {
-							java.lang.String s = input.readStringRequireUtf8();
-							bitField0_ |= 0x00000002;
-							name_ = s;
-							break;
-						}
-						default: {
-							if (!parseUnknownField(input, unknownFields, extensionRegistry, tag)) {
-								done = true;
-							}
-							break;
-						}
-					}
-				}
-			}
-			catch (com.google.protobuf.InvalidProtocolBufferException e) {
-				throw e.setUnfinishedMessage(this);
-			}
-			catch (com.google.protobuf.UninitializedMessageException e) {
-				throw e.asInvalidProtocolBufferException().setUnfinishedMessage(this);
-			}
-			catch (java.io.IOException e) {
-				throw new com.google.protobuf.InvalidProtocolBufferException(e).setUnfinishedMessage(this);
-			}
-			finally {
-				this.unknownFields = unknownFields.build();
-				makeExtensionsImmutable();
-			}
-		}
-
 		public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() {
-			return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_descriptor;
+			return org.springframework.pulsar.listener.Proto.internal_static_Person_descriptor;
 		}
 
 		@java.lang.Override
 		protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internalGetFieldAccessorTable() {
-			return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_fieldAccessorTable
+			return org.springframework.pulsar.listener.Proto.internal_static_Person_fieldAccessorTable
 				.ensureFieldAccessorsInitialized(org.springframework.pulsar.listener.Proto.Person.class,
 						org.springframework.pulsar.listener.Proto.Person.Builder.class);
 		}
 
-		private int bitField0_;
+		public static final int ID_FIELD_NUMBER = 2;
 
-		public static final int ID_FIELD_NUMBER = 1;
-
-		private int id_;
+		private int id_ = 0;
 
 		/**
-		 * <code>optional int32 id = 1;</code>
-		 * @return Whether the id field is set.
-		 */
-		@java.lang.Override
-		public boolean hasId() {
-			return ((bitField0_ & 0x00000001) != 0);
-		}
-
-		/**
-		 * <code>optional int32 id = 1;</code>
+		 * <code>int32 id = 2;</code>
 		 * @return The id.
 		 */
 		@java.lang.Override
@@ -168,21 +88,13 @@ public final class Proto {
 			return id_;
 		}
 
-		public static final int NAME_FIELD_NUMBER = 2;
+		public static final int NAME_FIELD_NUMBER = 1;
 
-		private volatile java.lang.Object name_;
+		@SuppressWarnings("serial")
+		private volatile java.lang.Object name_ = "";
 
 		/**
-		 * <code>optional string name = 2;</code>
-		 * @return Whether the name field is set.
-		 */
-		@java.lang.Override
-		public boolean hasName() {
-			return ((bitField0_ & 0x00000002) != 0);
-		}
-
-		/**
-		 * <code>optional string name = 2;</code>
+		 * <code>string name = 1;</code>
 		 * @return The name.
 		 */
 		@java.lang.Override
@@ -200,7 +112,7 @@ public final class Proto {
 		}
 
 		/**
-		 * <code>optional string name = 2;</code>
+		 * <code>string name = 1;</code>
 		 * @return The bytes for name.
 		 */
 		@java.lang.Override
@@ -232,13 +144,13 @@ public final class Proto {
 
 		@java.lang.Override
 		public void writeTo(com.google.protobuf.CodedOutputStream output) throws java.io.IOException {
-			if (((bitField0_ & 0x00000001) != 0)) {
-				output.writeInt32(1, id_);
+			if (!com.google.protobuf.GeneratedMessageV3.isStringEmpty(name_)) {
+				com.google.protobuf.GeneratedMessageV3.writeString(output, 1, name_);
 			}
-			if (((bitField0_ & 0x00000002) != 0)) {
-				com.google.protobuf.GeneratedMessageV3.writeString(output, 2, name_);
+			if (id_ != 0) {
+				output.writeInt32(2, id_);
 			}
-			unknownFields.writeTo(output);
+			getUnknownFields().writeTo(output);
 		}
 
 		@java.lang.Override
@@ -248,13 +160,13 @@ public final class Proto {
 				return size;
 
 			size = 0;
-			if (((bitField0_ & 0x00000001) != 0)) {
-				size += com.google.protobuf.CodedOutputStream.computeInt32Size(1, id_);
+			if (!com.google.protobuf.GeneratedMessageV3.isStringEmpty(name_)) {
+				size += com.google.protobuf.GeneratedMessageV3.computeStringSize(1, name_);
 			}
-			if (((bitField0_ & 0x00000002) != 0)) {
-				size += com.google.protobuf.GeneratedMessageV3.computeStringSize(2, name_);
+			if (id_ != 0) {
+				size += com.google.protobuf.CodedOutputStream.computeInt32Size(2, id_);
 			}
-			size += unknownFields.getSerializedSize();
+			size += getUnknownFields().getSerializedSize();
 			memoizedSize = size;
 			return size;
 		}
@@ -269,19 +181,11 @@ public final class Proto {
 			}
 			org.springframework.pulsar.listener.Proto.Person other = (org.springframework.pulsar.listener.Proto.Person) obj;
 
-			if (hasId() != other.hasId())
+			if (getId() != other.getId())
 				return false;
-			if (hasId()) {
-				if (getId() != other.getId())
-					return false;
-			}
-			if (hasName() != other.hasName())
+			if (!getName().equals(other.getName()))
 				return false;
-			if (hasName()) {
-				if (!getName().equals(other.getName()))
-					return false;
-			}
-			if (!unknownFields.equals(other.unknownFields))
+			if (!getUnknownFields().equals(other.getUnknownFields()))
 				return false;
 			return true;
 		}
@@ -293,15 +197,11 @@ public final class Proto {
 			}
 			int hash = 41;
 			hash = (19 * hash) + getDescriptor().hashCode();
-			if (hasId()) {
-				hash = (37 * hash) + ID_FIELD_NUMBER;
-				hash = (53 * hash) + getId();
-			}
-			if (hasName()) {
-				hash = (37 * hash) + NAME_FIELD_NUMBER;
-				hash = (53 * hash) + getName().hashCode();
-			}
-			hash = (29 * hash) + unknownFields.hashCode();
+			hash = (37 * hash) + ID_FIELD_NUMBER;
+			hash = (53 * hash) + getId();
+			hash = (37 * hash) + NAME_FIELD_NUMBER;
+			hash = (53 * hash) + getName().hashCode();
+			hash = (29 * hash) + getUnknownFields().hashCode();
 			memoizedHashCode = hash;
 			return hash;
 		}
@@ -396,19 +296,19 @@ public final class Proto {
 		}
 
 		/**
-		 * Protobuf type {@code proto.Person}
+		 * Protobuf type {@code Person}
 		 */
 		public static final class Builder extends com.google.protobuf.GeneratedMessageV3.Builder<Builder> implements
-				// @@protoc_insertion_point(builder_implements:proto.Person)
+				// @@protoc_insertion_point(builder_implements:Person)
 				org.springframework.pulsar.listener.Proto.PersonOrBuilder {
 
 			public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() {
-				return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_descriptor;
+				return org.springframework.pulsar.listener.Proto.internal_static_Person_descriptor;
 			}
 
 			@java.lang.Override
 			protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internalGetFieldAccessorTable() {
-				return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_fieldAccessorTable
+				return org.springframework.pulsar.listener.Proto.internal_static_Person_fieldAccessorTable
 					.ensureFieldAccessorsInitialized(org.springframework.pulsar.listener.Proto.Person.class,
 							org.springframework.pulsar.listener.Proto.Person.Builder.class);
 			}
@@ -416,32 +316,26 @@ public final class Proto {
 			// Construct using
 			// org.springframework.pulsar.listener.Proto.Person.newBuilder()
 			private Builder() {
-				maybeForceBuilderInitialization();
+
 			}
 
 			private Builder(com.google.protobuf.GeneratedMessageV3.BuilderParent parent) {
 				super(parent);
-				maybeForceBuilderInitialization();
-			}
 
-			private void maybeForceBuilderInitialization() {
-				if (com.google.protobuf.GeneratedMessageV3.alwaysUseFieldBuilders) {
-				}
 			}
 
 			@java.lang.Override
 			public Builder clear() {
 				super.clear();
+				bitField0_ = 0;
 				id_ = 0;
-				bitField0_ = (bitField0_ & ~0x00000001);
 				name_ = "";
-				bitField0_ = (bitField0_ & ~0x00000002);
 				return this;
 			}
 
 			@java.lang.Override
 			public com.google.protobuf.Descriptors.Descriptor getDescriptorForType() {
-				return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_descriptor;
+				return org.springframework.pulsar.listener.Proto.internal_static_Person_descriptor;
 			}
 
 			@java.lang.Override
@@ -462,21 +356,23 @@ public final class Proto {
 			public org.springframework.pulsar.listener.Proto.Person buildPartial() {
 				org.springframework.pulsar.listener.Proto.Person result = new org.springframework.pulsar.listener.Proto.Person(
 						this);
-				int from_bitField0_ = bitField0_;
-				int to_bitField0_ = 0;
-				if (((from_bitField0_ & 0x00000001) != 0)) {
-					result.id_ = id_;
-					to_bitField0_ |= 0x00000001;
+				if (bitField0_ != 0) {
+					buildPartial0(result);
 				}
-				if (((from_bitField0_ & 0x00000002) != 0)) {
-					to_bitField0_ |= 0x00000002;
-				}
-				result.name_ = name_;
-				result.bitField0_ = to_bitField0_;
 				onBuilt();
 				return result;
 			}
 
+			private void buildPartial0(org.springframework.pulsar.listener.Proto.Person result) {
+				int from_bitField0_ = bitField0_;
+				if (((from_bitField0_ & 0x00000001) != 0)) {
+					result.id_ = id_;
+				}
+				if (((from_bitField0_ & 0x00000002) != 0)) {
+					result.name_ = name_;
+				}
+			}
+
 			@java.lang.Override
 			public Builder clone() {
 				return super.clone();
@@ -523,15 +419,15 @@ public final class Proto {
 			public Builder mergeFrom(org.springframework.pulsar.listener.Proto.Person other) {
 				if (other == org.springframework.pulsar.listener.Proto.Person.getDefaultInstance())
 					return this;
-				if (other.hasId()) {
+				if (other.getId() != 0) {
 					setId(other.getId());
 				}
-				if (other.hasName()) {
-					bitField0_ |= 0x00000002;
+				if (!other.getName().isEmpty()) {
 					name_ = other.name_;
+					bitField0_ |= 0x00000002;
 					onChanged();
 				}
-				this.mergeUnknownFields(other.unknownFields);
+				this.mergeUnknownFields(other.getUnknownFields());
 				onChanged();
 				return this;
 			}
@@ -544,19 +440,42 @@ public final class Proto {
 			@java.lang.Override
 			public Builder mergeFrom(com.google.protobuf.CodedInputStream input,
 					com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException {
-				org.springframework.pulsar.listener.Proto.Person parsedMessage = null;
+				if (extensionRegistry == null) {
+					throw new java.lang.NullPointerException();
+				}
 				try {
-					parsedMessage = PARSER.parsePartialFrom(input, extensionRegistry);
+					boolean done = false;
+					while (!done) {
+						int tag = input.readTag();
+						switch (tag) {
+							case 0:
+								done = true;
+								break;
+							case 10: {
+								name_ = input.readStringRequireUtf8();
+								bitField0_ |= 0x00000002;
+								break;
+							} // case 10
+							case 16: {
+								id_ = input.readInt32();
+								bitField0_ |= 0x00000001;
+								break;
+							} // case 16
+							default: {
+								if (!super.parseUnknownField(input, extensionRegistry, tag)) {
+									done = true; // was an endgroup tag
+								}
+								break;
+							} // default:
+						} // switch (tag)
+					} // while (!done)
 				}
 				catch (com.google.protobuf.InvalidProtocolBufferException e) {
-					parsedMessage = (org.springframework.pulsar.listener.Proto.Person) e.getUnfinishedMessage();
 					throw e.unwrapIOException();
 				}
 				finally {
-					if (parsedMessage != null) {
-						mergeFrom(parsedMessage);
-					}
-				}
+					onChanged();
+				} // finally
 				return this;
 			}
 
@@ -565,16 +484,7 @@ public final class Proto {
 			private int id_;
 
 			/**
-			 * <code>optional int32 id = 1;</code>
-			 * @return Whether the id field is set.
-			 */
-			@java.lang.Override
-			public boolean hasId() {
-				return ((bitField0_ & 0x00000001) != 0);
-			}
-
-			/**
-			 * <code>optional int32 id = 1;</code>
+			 * <code>int32 id = 2;</code>
 			 * @return The id.
 			 */
 			@java.lang.Override
@@ -583,19 +493,20 @@ public final class Proto {
 			}
 
 			/**
-			 * <code>optional int32 id = 1;</code>
+			 * <code>int32 id = 2;</code>
 			 * @param value The id to set.
 			 * @return This builder for chaining.
 			 */
 			public Builder setId(int value) {
-				bitField0_ |= 0x00000001;
+
 				id_ = value;
+				bitField0_ |= 0x00000001;
 				onChanged();
 				return this;
 			}
 
 			/**
-			 * <code>optional int32 id = 1;</code>
+			 * <code>int32 id = 2;</code>
 			 * @return This builder for chaining.
 			 */
 			public Builder clearId() {
@@ -608,15 +519,7 @@ public final class Proto {
 			private java.lang.Object name_ = "";
 
 			/**
-			 * <code>optional string name = 2;</code>
-			 * @return Whether the name field is set.
-			 */
-			public boolean hasName() {
-				return ((bitField0_ & 0x00000002) != 0);
-			}
-
-			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @return The name.
 			 */
 			public java.lang.String getName() {
@@ -633,7 +536,7 @@ public final class Proto {
 			}
 
 			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @return The bytes for name.
 			 */
 			public com.google.protobuf.ByteString getNameBytes() {
@@ -650,7 +553,7 @@ public final class Proto {
 			}
 
 			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @param value The name to set.
 			 * @return This builder for chaining.
 			 */
@@ -658,25 +561,25 @@ public final class Proto {
 				if (value == null) {
 					throw new NullPointerException();
 				}
-				bitField0_ |= 0x00000002;
 				name_ = value;
+				bitField0_ |= 0x00000002;
 				onChanged();
 				return this;
 			}
 
 			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @return This builder for chaining.
 			 */
 			public Builder clearName() {
-				bitField0_ = (bitField0_ & ~0x00000002);
 				name_ = getDefaultInstance().getName();
+				bitField0_ = (bitField0_ & ~0x00000002);
 				onChanged();
 				return this;
 			}
 
 			/**
-			 * <code>optional string name = 2;</code>
+			 * <code>string name = 1;</code>
 			 * @param value The bytes for name to set.
 			 * @return This builder for chaining.
 			 */
@@ -685,8 +588,8 @@ public final class Proto {
 					throw new NullPointerException();
 				}
 				checkByteStringIsUtf8(value);
-				bitField0_ |= 0x00000002;
 				name_ = value;
+				bitField0_ |= 0x00000002;
 				onChanged();
 				return this;
 			}
@@ -701,11 +604,11 @@ public final class Proto {
 				return super.mergeUnknownFields(unknownFields);
 			}
 
-			// @@protoc_insertion_point(builder_scope:proto.Person)
+			// @@protoc_insertion_point(builder_scope:Person)
 
 		}
 
-		// @@protoc_insertion_point(class_scope:proto.Person)
+		// @@protoc_insertion_point(class_scope:Person)
 		private static final org.springframework.pulsar.listener.Proto.Person DEFAULT_INSTANCE;
 		static {
 			DEFAULT_INSTANCE = new org.springframework.pulsar.listener.Proto.Person();
@@ -720,7 +623,21 @@ public final class Proto {
 			public Person parsePartialFrom(com.google.protobuf.CodedInputStream input,
 					com.google.protobuf.ExtensionRegistryLite extensionRegistry)
 					throws com.google.protobuf.InvalidProtocolBufferException {
-				return new Person(input, extensionRegistry);
+				Builder builder = newBuilder();
+				try {
+					builder.mergeFrom(input, extensionRegistry);
+				}
+				catch (com.google.protobuf.InvalidProtocolBufferException e) {
+					throw e.setUnfinishedMessage(builder.buildPartial());
+				}
+				catch (com.google.protobuf.UninitializedMessageException e) {
+					throw e.asInvalidProtocolBufferException().setUnfinishedMessage(builder.buildPartial());
+				}
+				catch (java.io.IOException e) {
+					throw new com.google.protobuf.InvalidProtocolBufferException(e)
+						.setUnfinishedMessage(builder.buildPartial());
+				}
+				return builder.buildPartial();
 			}
 		};
 
@@ -740,9 +657,9 @@ public final class Proto {
 
 	}
 
-	private static final com.google.protobuf.Descriptors.Descriptor internal_static_proto_Person_descriptor;
+	private static final com.google.protobuf.Descriptors.Descriptor internal_static_Person_descriptor;
 
-	private static final com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internal_static_proto_Person_fieldAccessorTable;
+	private static final com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internal_static_Person_fieldAccessorTable;
 
 	public static com.google.protobuf.Descriptors.FileDescriptor getDescriptor() {
 		return descriptor;
@@ -750,14 +667,14 @@ public final class Proto {
 
 	private static com.google.protobuf.Descriptors.FileDescriptor descriptor;
 	static {
-		java.lang.String[] descriptorData = { "\n\014person.proto\022\005proto\"<\n\006Person\022\017\n\002id\030\001 "
-				+ "\001(\005H\000\210\001\001\022\021\n\004name\030\002 \001(\tH\001\210\001\001B\005\n\003_idB\007\n\005_n"
-				+ "ameB,\n#org.springframework.pulsar.listen" + "erB\005Protob\006proto3" };
+		java.lang.String[] descriptorData = {
+				"\n\014person.proto\"\"\n\006Person\022\n\n\002id\030\002 \001(\005\022\014\n\004"
+						+ "name\030\001 \001(\tB,\n#org.springframework.pulsar" + ".listenerB\005Protob\006proto3" };
 		descriptor = com.google.protobuf.Descriptors.FileDescriptor.internalBuildGeneratedFileFrom(descriptorData,
 				new com.google.protobuf.Descriptors.FileDescriptor[] {});
-		internal_static_proto_Person_descriptor = getDescriptor().getMessageTypes().get(0);
-		internal_static_proto_Person_fieldAccessorTable = new com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
-				internal_static_proto_Person_descriptor, new java.lang.String[] { "Id", "Name", "Id", "Name", });
+		internal_static_Person_descriptor = getDescriptor().getMessageTypes().get(0);
+		internal_static_Person_fieldAccessorTable = new com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
+				internal_static_Person_descriptor, new java.lang.String[] { "Id", "Name", });
 	}
 
 	// @@protoc_insertion_point(outer_class_scope)
diff --git a/spring-pulsar/src/test/proto/README.adoc b/spring-pulsar/src/test/proto/README.adoc
new file mode 100644
index 00000000..017184de
--- /dev/null
+++ b/spring-pulsar/src/test/proto/README.adoc
@@ -0,0 +1,60 @@
+= How to regenerate the protos
+
+Every so often the generated Proto java POJO used for testing need to be updated.
+
+They are used in both `spring-pulsar` and `spring-pulsar-reactive`.
+
+== Make sure proper protoc installed
+
+Run the following command:
+----
+protoc --version
+----
+
+If the command is not found or returns version below `25.6` then you will need to install as follows.
+
+----
+wget -O protoc.zip https://github.com/protocolbuffers/protobuf/releases/download/v25.6/protoc-25.6-osx-aarch_64.zip
+
+mkdir tmp
+mv protoc.zip tmp
+cd tmp
+unzip protoc.zip
+mv bin/protoc /usr/local/bin
+cd ..
+rm -rf tmp
+----
+
+Check the version again:
+----
+protoc --version
+----
+Should now return version `25.6`. If it does not, figure it out.
+
+TIP: If you are on Mac you may see “macOS cannot verify that this app is free from malware”. If so, navigate to the binary and hold down "CTRL" and right-click and select "Open".
+
+== Regenerate the protos
+
+From the project root directory run the following commands:
+
+.spring-pulsar
+----
+pushd spring-pulsar/src/test/proto  > /dev/null 
+protoc --java_out=. person.proto
+cp -R org ../../../src/test/java
+rm -rf org
+popd > /dev/null
+./gradlew format
+----
+
+.spring-pulsar-reactive
+----
+pushd spring-pulsar-reactive/src/test/proto  > /dev/null
+protoc --java_out=. person.proto
+cp -R org ../../../src/test/java
+rm -rf org
+popd > /dev/null
+./gradlew format
+----
+
+Check in the modified generated POJOs
diff --git a/spring-pulsar/src/test/proto/person.proto b/spring-pulsar/src/test/proto/person.proto
new file mode 100644
index 00000000..72b63028
--- /dev/null
+++ b/spring-pulsar/src/test/proto/person.proto
@@ -0,0 +1,9 @@
+syntax = "proto3";
+
+option java_package = "org.springframework.pulsar.listener";
+option java_outer_classname = "Proto";
+
+message Person {
+  int32 id = 2;
+  string name = 1;
+}