[Schema] Fix autoConsumeSchema deadLetter problem. by congbobo184 · Pull Request #9970 · apache/pulsar

congbobo184 · 2021-03-19T03:19:06Z

Motivation

in order to fix autoConsumeSchema deadLetter problem.
fix #9935

implement

Add AutoProduceSchema to fix it

Verifying this change

Add the tests for it

Does this pull request potentially affect one of the following parts:
If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

eolivelli

awesome work @congbobo184

I see that this new class is very similar to AutoProduceBytesSchema

is there a way to consolidate the logic in a common base class

eolivelli · 2021-03-19T07:25:45Z

+
+    @Override
+    public SchemaInfo getSchemaInfo() {
+        return BytesSchema.of().getSchemaInfo();


what about this version ?
IMHO we should delegate to the wrapped schema

@Override public SchemaInfo getSchemaInfo() { ensureSchemaInitialized(); return schema.getSchemaInfo(); }

congbobo184 · 2021-03-19T14:23:29Z

@eolivelli thanks for you comment, I change the deadLetterProducer to AutoProduceBytesSchema. please review again.

eolivelli · 2021-03-22T14:17:15Z

+    public CompletableFuture<SchemaInfo> getSchemaInfo(byte[] schemaVersion) {
+        ensureSchemaInitialized();
+        if (schema.supportSchemaVersioning()) {
+            if (schemaVersion == null) {


isn't is better to consistently return an error if schemaVersion is null ?

eolivelli · 2021-03-22T14:21:22Z

        return this.schema;
    }

+    public CompletableFuture<Schema<?>> getOriginalSchema() {


can you please add a javadoc that explains why method returns a CompletableFuture ?
for instance we could need to download the schema from the network.

can we name this method "fetchSchema" ?
This way the name itself reads that this is not a simple "getter" but that the operation may perform additional operations.

eolivelli

are you sure that you are not adding another patch on top of this patch ?
I see changed related to Pulsar IO and also to PulsarProducer that look unrelated

eolivelli · 2021-03-23T14:42:46Z

@@ -88,15 +88,17 @@ public class PulsarSink<T> implements Sink<T> {

    private interface PulsarSinkProcessor<T> {


why are we changing PulsarSink ?

PulsarSink use MessageImpl.getSchema()

eolivelli · 2021-03-23T14:43:07Z

            return;
        }

-        if (!populateMessageSchema(msg, callback)) {


is it necessary to change ProducerImpl ?

same as above.

eolivelli · 2021-03-23T14:43:40Z

    }

-    default Schema<T> getSchema() {
+    default Schema getSchema() {


why are we removing this template parameter ?
it is wrong, if the Record is of type T, than the Schema is for type T

i will change it.

eolivelli · 2021-03-23T14:44:01Z

@@ -128,31 +130,43 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws

    @Override
    public void received(Consumer<T> consumer, Message<T> message) {


why do we need to change PulsarSource ?

# Conflicts: # pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java # pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java # pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java # pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java # pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java # pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java # pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AbstractStructSchema.java # pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java # pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/SchemaHash.java # pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java # pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/sink/PulsarSinkTest.java # pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/PulsarSourceTest.java

congbobo184 · 2021-06-02T09:54:26Z

@eolivelli please review again thanks.

eolivelli

Great!
I am happy the getReaderSchema helped to fix this problem

* [Schema] Fix autoConsumeSchema deadLetter problem. * Change the schema to AutoProduceBytesSchema * Fix the message getSchemaInfo * add message getOriginalSchema * Change message getSchema method to async * Fix the test Co-authored-by: congbo <congbobo184@github.com>

…14317) If a message reached maxRedeliverCount, it will send to deadLetterTopic, since 2.8.0, this mechanism is broken, it was introduced in #9970

…14317) If a message reached maxRedeliverCount, it will send to deadLetterTopic, since 2.8.0, this mechanism is broken, it was introduced in #9970 (cherry picked from commit 16beb9d)

mattisonchao · 2022-03-01T00:05:21Z

            List<MessageImpl<T>> finalDeadLetterMessages = deadLetterMessages;
            MessageIdImpl finalMessageId = messageId;
-            deadLetterProducer.thenAccept(producerDLQ -> {
+            deadLetterProducer.thenAcceptAsync(producerDLQ -> {


@congbobo184 Why do we use a common pool to run the next logic?

…pache#14317) If a message reached maxRedeliverCount, it will send to deadLetterTopic, since 2.8.0, this mechanism is broken, it was introduced in apache#9970

…14317) If a message reached maxRedeliverCount, it will send to deadLetterTopic, since 2.8.0, this mechanism is broken, it was introduced in #9970 (cherry picked from commit 16beb9d)

* [Schema] Fix autoConsumeSchema deadLetter problem. * Change the schema to AutoProduceBytesSchema * Fix the message getSchemaInfo * add message getOriginalSchema * Change message getSchema method to async * Fix the test Co-authored-by: congbo <congbobo184@github.com>

…pache#14317) If a message reached maxRedeliverCount, it will send to deadLetterTopic, since 2.8.0, this mechanism is broken, it was introduced in apache#9970

[Schema] Fix autoConsumeSchema deadLetter problem.

7322969

eolivelli requested changes Mar 19, 2021

View reviewed changes

Change the schema to AutoProduceBytesSchema

e63bc6f

congbo added 2 commits March 22, 2021 22:06

Fix the message getSchemaInfo

67cec47

add message getOriginalSchema

d9d0cab

eolivelli requested changes Mar 22, 2021

View reviewed changes

congbo added 2 commits March 23, 2021 17:08

Change message getSchema method to async

9b9a67c

Fix the test

a15a5ae

eolivelli requested changes Mar 23, 2021

View reviewed changes

codelipenghui added this to the 2.8.0 milestone Apr 2, 2021

codelipenghui assigned congbobo184 Apr 2, 2021

codelipenghui modified the milestones: 2.8.0, 2.9.0 May 21, 2021

congbobo184 requested review from codelipenghui and gaoran10 June 2, 2021 09:54

congbobo184 added the type/bug The PR fixed a bug or issue reported a bug label Jun 2, 2021

eolivelli approved these changes Jun 2, 2021

View reviewed changes

eolivelli merged commit 9b28bf3 into apache:master Jun 2, 2021

congbobo184 modified the milestones: 2.9.0, 2.8.0 Jun 2, 2021

Shawyeok mentioned this pull request Feb 16, 2022

[pulsar-client] Fix send to deadLetterTopic not working when reach maxRedeliverCount #14317

Merged

3 tasks

mattisonchao reviewed Mar 1, 2022

View reviewed changes

congbobo184 deleted the congbobo184_fix_auto_consume_schema_problem branch March 24, 2022 04:58

shibd mentioned this pull request Dec 6, 2022

[feat] Support Dead Letter Topic. apache/pulsar-client-cpp#139

Merged

4 tasks

		@@ -88,15 +88,17 @@ public class PulsarSink<T> implements Sink<T> {

		private interface PulsarSinkProcessor<T> {

		@@ -128,31 +130,43 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws

		@Override
		public void received(Consumer<T> consumer, Message<T> message) {

Uh oh!

Conversation

congbobo184 commented Mar 19, 2021

Motivation

implement

Verifying this change

Uh oh!

eolivelli left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

congbobo184 commented Mar 19, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

eolivelli left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

congbobo184 commented Jun 2, 2021

Uh oh!

eolivelli left a comment

Choose a reason for hiding this comment

Uh oh!

mattisonchao Mar 1, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

mattisonchao Mar 1, 2022 •

edited

Loading