feat: consumer schema validation for avro by ccm32004 · Pull Request #59 · numaproj-contrib/apache-pulsar-java

ccm32004 · 2026-02-18T16:11:54Z

when consumer receives messages from topic that is registered with an avro schema, will perform validation to ensure that correct fields are defined in message payload, and that correct format is used (not json etc). uses apache pulsar client consumer AUTO_CONSUME function.

added configuration behaviour:
- user also has option to turn auto consume on or off for enable/disable consumer schema validation
added tests to verify behaviour

Testing notes:

tested with pulsar streamnative cluster instance
created topic registered with avro schema of only one field (ie name field), published messages to that topic with one field, then reregistered the same topic with a different avro schema of two fields
created consumer pipeline connecting to pulsar topic, consumer pulls pulsar topic schema with two fields, but receives avro messages with only one field, thus performs decoding on message payload and fails

Figure 1: application fails after pipeline consumes methods that don't match schema registered to topic

Tested happy path where messages matched topic schema as well, verified that correct message logs were being consumed, also sink shows malformed json because it cannot deserialize avro bytes

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

src/main/java/io/numaproj/pulsar/config/consumer/PulsarConsumerProperties.java

src/main/java/io/numaproj/pulsar/consumer/GenericRecordToBytes.java

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

src/main/java/io/numaproj/pulsar/consumer/PulsarConsumerManager.java

src/main/java/io/numaproj/pulsar/consumer/PulsarSource.java

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

KeranYang · 2026-02-23T17:41:26Z

src/main/java/io/numaproj/pulsar/consumer/PulsarConsumerManager.java

+        bytesConsumer = pulsarClient.newConsumer(Schema.BYTES)
+                .loadConf(pulsarConsumerProperties.getConsumerConfig())
+                .batchReceivePolicy(batchPolicy)
+                .subscriptionType(SubscriptionType.Shared)


Suggested change

.subscriptionType(SubscriptionType.Shared)

.subscriptionType(SubscriptionType.Shared) // Must be shared to support multiple pods

KeranYang · 2026-02-23T17:41:44Z

src/main/java/io/numaproj/pulsar/consumer/PulsarConsumerManager.java

                .loadConf(pulsarConsumerProperties.getConsumerConfig())
                .batchReceivePolicy(batchPolicy)
-                .subscriptionType(SubscriptionType.Shared) // Must be shared to support multiple pods
+                .subscriptionType(SubscriptionType.Shared)


Suggested change

.subscriptionType(SubscriptionType.Shared)

.subscriptionType(SubscriptionType.Shared) // Must be shared to support multiple pods

KeranYang · 2026-02-23T17:48:11Z

src/main/java/io/numaproj/pulsar/consumer/PulsarConsumerManager.java

+     * Returns the consumer used for receiving and acknowledgment, creating it if necessary.
+     * Matches useAutoConsumeSchema. Never null.
+     */
+    public Consumer<?> getConsumer() throws PulsarClientException {


There are two methods caller can call to get a byte array consumer. Please keep one. See if we can mark getOrCreateGenericRecordConsumer and getOrCreateGenericRecordConsumer as private.

KeranYang · 2026-02-23T18:00:12Z

src/main/java/io/numaproj/pulsar/consumer/PulsarSource.java

+            if (t instanceof SchemaSerializationException
+                    || t instanceof IOException
+                    || t instanceof UnsupportedOperationException) {


Please document more details about each of these exceptions. When will it be thrown under a schema validation case.

src/main/java/io/numaproj/pulsar/consumer/PulsarSource.java

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

KeranYang · 2026-02-25T15:38:18Z

src/main/java/io/numaproj/pulsar/consumer/PulsarConsumerManager.java

                .maxNumMessages((int) count)
-                .timeout((int) timeoutMillis, TimeUnit.MILLISECONDS) // We do not expect user to specify a number larger
-                                                                     // than 2^63 - 1 which will cause an overflow
+                .timeout((int) timeoutMillis, TimeUnit.MILLISECONDS)


Please keep the original comments.

KeranYang · 2026-02-25T15:43:52Z

src/test/java/io/numaproj/pulsar/consumer/PulsarSourceTest.java

            consumerManagerMock = mock(PulsarConsumerManager.class);
            consumerMock = mock(Consumer.class);
-            // Inject the mocked PulsarConsumerManager into pulsarSource using
+             // Inject the mocked PulsarConsumerManager into pulsarSource using


Suggested change

// Inject the mocked PulsarConsumerManager into pulsarSource using

// Inject the mocked PulsarConsumerManager into pulsarSource using

Consider configuring Format on Save.

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

ccm32004 added 5 commits February 18, 2026 07:27

curr working ver

629ee1b

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

fixed/added tests

57b484b

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

refactor

797fa8d

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

fixed comments

27783aa

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

fixed test

033ea2a

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

ccm32004 requested a review from KeranYang February 18, 2026 16:19

KeranYang reviewed Feb 18, 2026

View reviewed changes

ccm32004 marked this pull request as draft February 18, 2026 17:29

ccm32004 added 2 commits February 19, 2026 10:21

rm generic method to bytes + created two consumers

e75a687

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

fix comments

86ecc32

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

ccm32004 marked this pull request as ready for review February 19, 2026 16:02

KeranYang reviewed Feb 19, 2026

View reviewed changes

ccm32004 added 2 commits February 19, 2026 16:35

address PR comments

ca0c613

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

fixed tests

328f859

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

KeranYang reviewed Feb 23, 2026

View reviewed changes

ccm32004 added 3 commits February 24, 2026 14:59

address some PR comments

854a0fe

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

fix test

590c214

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

edited comment

523fdd5

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

ccm32004 requested a review from KeranYang February 24, 2026 20:38

KeranYang reviewed Feb 25, 2026

View reviewed changes

fix formatting/comments

6b3a2ab

Signed-off-by: Cece Ma <mayuqing131@gmail.com>

ccm32004 requested a review from KeranYang February 25, 2026 20:56

KeranYang approved these changes Feb 25, 2026

View reviewed changes

Merge branch 'main' into feature/consumer-schema-validation

68d7949

KeranYang merged commit 97d782f into main Feb 25, 2026
2 checks passed

KeranYang deleted the feature/consumer-schema-validation branch February 25, 2026 21:53

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: consumer schema validation for avro#59

feat: consumer schema validation for avro#59
KeranYang merged 14 commits intomainfrom
feature/consumer-schema-validation

ccm32004 commented Feb 18, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

KeranYang Feb 23, 2026

Uh oh!

KeranYang Feb 23, 2026

Uh oh!

KeranYang Feb 23, 2026

Uh oh!

KeranYang Feb 23, 2026

Uh oh!

Uh oh!

KeranYang Feb 25, 2026

Uh oh!

KeranYang Feb 25, 2026

Uh oh!

KeranYang Feb 25, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

	.subscriptionType(SubscriptionType.Shared)
	.subscriptionType(SubscriptionType.Shared) // Must be shared to support multiple pods

	// Inject the mocked PulsarConsumerManager into pulsarSource using
	// Inject the mocked PulsarConsumerManager into pulsarSource using

Conversation

ccm32004 commented Feb 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

KeranYang Feb 23, 2026

Choose a reason for hiding this comment

Uh oh!

KeranYang Feb 23, 2026

Choose a reason for hiding this comment

Uh oh!

KeranYang Feb 23, 2026

Choose a reason for hiding this comment

Uh oh!

KeranYang Feb 23, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

KeranYang Feb 25, 2026

Choose a reason for hiding this comment

Uh oh!

KeranYang Feb 25, 2026

Choose a reason for hiding this comment

Uh oh!

KeranYang Feb 25, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ccm32004 commented Feb 18, 2026 •

edited

Loading