[fix][broker] Fix schema does not replicate successfully #17049

codelipenghui · 2022-08-10T16:45:24Z

Motivation

#11441 supports replicate schema to remote clusters.
But there is a mistake that the returned schema state is incorrect.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java

Lines 765 to 770 in e826d84

    
           SchemaState getSchemaState() { 
        
               if (getSchemaInfo() == null) { 
        
                   return SchemaState.Ready; 
        
               } 
        
               return schemaState; 
        
           }

Because the replicator used MessageImpl will not have the schema.
And this will cause the producer to skip the schema upload.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

Lines 2147 to 2149 in e826d84

    
           if (op.msg != null && op.msg.getSchemaState() == None) { 
        
               tryRegisterSchema(cnx, op.msg, op.callback, this.connectionHandler.getEpoch()); 
        
               return;

We should remove

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java

Lines 766 to 768 in e826d84

    
           if (getSchemaInfo() == null) { 
        
               return SchemaState.Ready; 
        
           }

To return the correct schema state.

And then we should also provide the correct schema hash.
If the message is used by the replicator, the schema hash should
be based on the replicator schema. Otherwise, it should use based
on the schema of the message.

Modification

Fixed the incorrect returned schema state
Provide the method for getting schema hash for MessageImpl

Verification

Update the test only to create a producer to one cluster.
Because if create a producer for other clusters, the producer will
upload the schema. This is the reason why the test can get
passed before.

doc-not-needed

### Motivation apache#11441 supports replicate schema to remote clusters. But there is a mistake that the returned schema state is incorrect. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L765-L770 Because the replicator used MessageImpl will not have the schema. And this will cause the producer to skip the schema upload. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2147-L2149 We should remove https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L766-L768 To return the correct schema state. And then we should also provide the correct schema hash. If the message is used by the replicator, the schema hash should be based on the replicator schema. Otherwise, it should use based on the schema of the message. ### Modification - Fixed the incorrect returned schema state - Provide the method for getting schema hash for MessageImpl ### Verification Update the test only to create producer to one cluster. Because if create a producer for other clusters, the producer will upload the schema. This is the reason that why the test can get pass before.

mattisonchao

LGTM

gaoran10

LGTM

HQebupt · 2022-08-13T03:17:28Z

/pulsarbot rerun-failure-checks

codelipenghui · 2022-08-16T16:25:21Z

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java

-        Position deletedPos = cursor.getMarkDeletedPosition();
-        Position readPos = cursor.getReadPosition();
-
-        Awaitility.await().timeout(30, TimeUnit.SECONDS).until(
-                () -> cursor.getMarkDeletedPosition().getEntryId() != (cursor.getReadPosition().getEntryId() - 1));
-
-        assertNotEquals((readPos.getEntryId() - 1), deletedPos.getEntryId());


Removed here because the producer might be closed with pending messages. After the cursor change to the close state, the producer will also be closed after a read entries failure. So that all the pending messages will fail with the Producer Already Closed exception, then the cursor will rewind

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java

Line 425 in 9f40cc1

replicator.cursor.rewind();

This will make the check failed.

codelipenghui · 2022-08-16T16:27:52Z

There should be another issue related to this part

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java

Line 367 in 9f40cc1

getSchemaInfo(msg).thenAccept(schemaInfo -> {

Here might break the message replication order. I'm working on a fix for now.
We should cherry-pick this one along with the ordering issue fix PR.

codelipenghui · 2022-08-17T01:34:18Z

/pulsarbot run-failure-checks

This reverts commit ee1563f.

### Motivation #11441 supports replicate schema to remote clusters. But there is a mistake that the returned schema state is incorrect. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L765-L770 Because the replicator used MessageImpl will not have the schema. And this will cause the producer to skip the schema upload. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2147-L2149 We should remove https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L766-L768 To return the correct schema state. And then we should also provide the correct schema hash. If the message is used by the replicator, the schema hash should be based on the replicator schema. Otherwise, it should use based on the schema of the message. ### Modification - Fixed the incorrect returned schema state - Provide the method for getting schema hash for MessageImpl

But there is a mistake that the returned schema state is incorrect. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L765-L770 Because the replicator used MessageImpl will not have the schema. And this will cause the producer to skip the schema upload. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2147-L2149 We should remove https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L766-L768 To return the correct schema state. And then we should also provide the correct schema hash. If the message is used by the replicator, the schema hash should be based on the replicator schema. Otherwise, it should use based on the schema of the message. - Fixed the incorrect returned schema state - Provide the method for getting schema hash for MessageImpl (cherry picked from commit 7689133)

But there is a mistake that the returned schema state is incorrect. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L765-L770 Because the replicator used MessageImpl will not have the schema. And this will cause the producer to skip the schema upload. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2147-L2149 We should remove https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L766-L768 To return the correct schema state. And then we should also provide the correct schema hash. If the message is used by the replicator, the schema hash should be based on the replicator schema. Otherwise, it should use based on the schema of the message. - Fixed the incorrect returned schema state - Provide the method for getting schema hash for MessageImpl (cherry picked from commit 7689133) (cherry picked from commit c1d305b)

But there is a mistake that the returned schema state is incorrect. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L765-L770 Because the replicator used MessageImpl will not have the schema. And this will cause the producer to skip the schema upload. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2147-L2149 We should remove https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L766-L768 To return the correct schema state. And then we should also provide the correct schema hash. If the message is used by the replicator, the schema hash should be based on the replicator schema. Otherwise, it should use based on the schema of the message. - Fixed the incorrect returned schema state - Provide the method for getting schema hash for MessageImpl (cherry picked from commit 7689133)

lhotari · 2022-10-05T04:02:27Z

build/run_unit_group.sh

@@ -134,7 +134,7 @@ function test_group_other() {
           -Dexclude='**/ManagedLedgerTest.java,
                   **/OffloadersCacheTest.java
                  **/PrimitiveSchemaTest.java,
-                  BlobStoreManagedLedgerOffloaderTest.java'
+                  BlobStoreManagedLedgerOffloaderTest.java' -DtestReuseFork=false


@codelipenghui do you remember why this was needed? I added #17943 to remove it.

The motivation is here #17152

Although I think it's safe to add it back

codelipenghui self-assigned this Aug 10, 2022

codelipenghui added doc-not-needed Your PR changes do not impact docs release/2.10.2 release/2.9.4 release/2.11.1 labels Aug 10, 2022

codelipenghui added this to the 2.12.0 milestone Aug 10, 2022

codelipenghui added type/bug The PR fixed a bug or issue reported a bug area/broker labels Aug 10, 2022

codelipenghui marked this pull request as ready for review August 10, 2022 16:46

codelipenghui added 2 commits August 11, 2022 11:44

Fix NPE

691fd25

Fix NPE

a756613

Technoboy- approved these changes Aug 11, 2022

View reviewed changes

shibd approved these changes Aug 11, 2022

View reviewed changes

mattisonchao approved these changes Aug 11, 2022

View reviewed changes

gaoran10 approved these changes Aug 12, 2022

View reviewed changes

HQebupt approved these changes Aug 13, 2022

View reviewed changes

coderzc approved these changes Aug 13, 2022

View reviewed changes

Fix Test

a995883

codelipenghui force-pushed the penghui/fix-replicate-schema branch from 23a7e95 to a995883 Compare August 16, 2022 16:21

codelipenghui commented Aug 16, 2022

View reviewed changes

codelipenghui added 3 commits August 17, 2022 16:48

Fix Test

ee1563f

Revert "Fix Test"

221e74b

This reverts commit ee1563f.

Disable reuseFork

312b4d5

Technoboy- mentioned this pull request Aug 17, 2022

[fix][flaky-test] ReplicatorTest.testReplicatorWithFailedAck #17144

Closed

4 tasks

Technoboy- added 2 commits August 18, 2022 09:48

fix test.

98eb1d4

fix param

0e3d62b

fix test.

cdd89cf

codelipenghui merged commit 7689133 into apache:master Aug 18, 2022

codelipenghui deleted the penghui/fix-replicate-schema branch August 18, 2022 03:22

This was referenced Aug 18, 2022

Flaky-test: ReplicatorTest.testReplicatorWithFailedAck #16500

Closed

Flaky-test: ConnectionTimeoutTest.testLowTimeout #15764

Closed

Technoboy- added the cherry-picked/branch-2.11 label Aug 18, 2022

Technoboy- mentioned this pull request Aug 18, 2022

[fix][tests] Fix the flaky test MultiTopicsConsumerImplTest.testParallelSubscribeAsync #17146

Closed

1 task

codelipenghui added cherry-picked/branch-2.10 and removed release/2.11.1 labels Aug 19, 2022

codelipenghui modified the milestones: 2.12.0, 2.11.0 Aug 19, 2022

codelipenghui mentioned this pull request Sep 7, 2022

[fix][broker] Fix the replicator unnecessary get schema request for BYTES schema #17523

Merged

4 tasks

codelipenghui added cherry-picked/branch-2.9 Archived: 2.9 is end of life and removed cherry-picked/branch-2.11 labels Sep 8, 2022

AnonHxy mentioned this pull request Oct 4, 2022

[Bug] A significant performance regression about producer publish throughput on master branch #17931

Closed

2 tasks

sijie mentioned this pull request Oct 4, 2022

ISSUE-17931: [Bug] A significant performance regression about producer publish throughput on master branch streamnative/pulsar-archived#4959

Open

2 tasks

lhotari reviewed Oct 5, 2022

View reviewed changes

AnonHxy mentioned this pull request Oct 6, 2022

[improve][client] Refactor SchemaHash to reduce call of hashFunction in SchemaHash #17948

Merged

5 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][broker] Fix schema does not replicate successfully #17049

[fix][broker] Fix schema does not replicate successfully #17049

codelipenghui commented Aug 10, 2022 •

edited by github-actions bot

Loading

mattisonchao left a comment

gaoran10 left a comment

HQebupt commented Aug 13, 2022

codelipenghui Aug 16, 2022

codelipenghui commented Aug 16, 2022

codelipenghui commented Aug 17, 2022

lhotari Oct 5, 2022

nicoloboschi Oct 5, 2022

	SchemaState getSchemaState() {
	if (getSchemaInfo() == null) {
	return SchemaState.Ready;
	}
	return schemaState;
	}

	if (op.msg != null && op.msg.getSchemaState() == None) {
	tryRegisterSchema(cnx, op.msg, op.callback, this.connectionHandler.getEpoch());
	return;

[fix][broker] Fix schema does not replicate successfully #17049

[fix][broker] Fix schema does not replicate successfully #17049

Conversation

codelipenghui commented Aug 10, 2022 • edited by github-actions bot Loading

Motivation

Modification

Verification

mattisonchao left a comment

Choose a reason for hiding this comment

gaoran10 left a comment

Choose a reason for hiding this comment

HQebupt commented Aug 13, 2022

codelipenghui Aug 16, 2022

Choose a reason for hiding this comment

codelipenghui commented Aug 16, 2022

codelipenghui commented Aug 17, 2022

lhotari Oct 5, 2022

Choose a reason for hiding this comment

nicoloboschi Oct 5, 2022

Choose a reason for hiding this comment

codelipenghui commented Aug 10, 2022 •

edited by github-actions bot

Loading