Fix schema not added when subscribe an empty topic without schema

[BewareMyPower]

Fixes #7728

Motivation

When a consumer with a schema subscribes an empty topic without schema, the current check uses isActive, which only checks if the topic can be deleted. However, it should check if there's any connected producer or consumer of this topic. For current implementation, even if a topic has no active producers of consumers, the topic's subscription list may be not empty and isActive will return true. Then the schema of consumer won't be attached to the topic and it will throw an IncompatibleSchemaException.

Modifications

• Check if the topic has active producers or consumers instead of whether it can be deleted.
• Add related tests.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change added tests testAutoCreatedSchema.

[BewareMyPower]

Update: this PR is compatible with #2669 now.

---

I think this change may need a discussion since it's not compatible with #2669.

@ivankelly @congbobo184 @codelipenghui

[BewareMyPower]

/pulsarbot run-failure-checks

[codelipenghui]

@BewareMyPower I think the root cause is the producer has written data into the topic but the schema does not upload? From the perspective of schema compatibility, if the topic has data but does not have a schema, this means the schema is bytes schema(Currently bytes schema is not uploaded), so I think the right fix is to trade as The topic with bytes schema, we should check the schema compatibility of the new producers/consumers.

Currently, the producer side also has a problem is a producer produce some message with bytes schema, then a new producer can publish messages with other schemas(e.g. Schema.INT64), this will introduce schema compatibility issues. Imagine that if a consumer consumes messages from the earliest position with Schema.INT64, will pass the schema compatibility check but will fail when deserializing messages.

[BewareMyPower]

@codelipenghui

if the topic has data but does not have a schema

I agree, so even if hasSchema is false, if ledger.getTotalSize() != 0, we should still go to checkSchemaCompatibleForConsumer.

However, I don't understand why we also need the check of isActive(InactiveTopicDeleteMode.delete_when_no_subscriptions). It causes the failure of testAutoCreatedSchema before applying changes of this PR since the topic has no data and it shouldn't fail. Does changing the conditional check to if (hasSchema || (ledger.getTotalSize() != 0)) have any problem?

[codelipenghui]

Does changing the conditional check to if (hasSchema || (ledger.getTotalSize() != 0)) have any problem?

We need to check if there are producers or consumers connect to this topic. Otherwise, if the old producers and consumers connect with bytes schema, then we change the schema type to schema.AVRO, this will break the old producers and consumers.

Check no subscriptions is equals to no consumers.

[BewareMyPower]

We need to check if there are producers or consumers connect to this topic.

Check no subscriptions is equals to no consumers.

So even the consumer has closed, it's still treated as connected? Take my test as the example

        try (Consumer<byte[]> consumer
                     = pulsarClient.newConsumer(Schema.BYTES).topic(topic2).subscriptionName("sub").subscribe()) {
        }
        try (Consumer<String> consumer
                     = pulsarClient.newConsumer(Schema.STRING).topic(topic2).subscriptionName("sub").subscribe()) {
        }

During the ServerCnx#handleSubscribe for the second consumer, the subscriptions is not empty. However the first consumer has already closed.

[codelipenghui]

If all consumers disconnected from a subscription, I think the subscription will remove from the topic? If not, we should check consumer size, This will be more clear.

[BewareMyPower]

update: The integration test is just a flaky test.

---

There's still a test failure in testPrimitiveSchemaTypeCompatibilityCheck, and the number of producers should also be checked here. I'll fix it soon.

[BewareMyPower]

/pulsarbot run-failure-checks

[BewareMyPower]

/pulsarbot run-failure-checks

[codelipenghui]

The change looks good to me. I think the producer side also has a similar problem? If there are producers or consumers connect to the topic with bytes schema, then when a new producer connect with avro schema, I think the schema of the topic will change to avro. This will break the existing producers/consumers, could you please also help push a separate PR to fix it?

[codelipenghui]

@congbobo184 @gaoran10 @limingnihao Could you please also help review this PR? You are working on the Pulsar schema currently, make sure we are on the same page.

[eolivelli]

Very good.
I left a little suggestion.

Apart from that, can you please add the same test for non persistent topics and ensure that the behaviour is consistent?

[congbobo184]

LGTM!

[BewareMyPower]

I've fixed the same issue for non-persistent topics, PTAL @eolivelli

[BewareMyPower]

I think the producer side also has a similar problem? If there are producers or consumers connect to the topic with bytes schema, then when a new producer connect with avro schema, I think the schema of the topic will change to avro.

@codelipenghui I'm not sure if it's the intended design for producer. Currently handleProducer uses tryAddSchema that is introduced in PIP-43 and #5165, while handleSubscribe uses addSchemaIfIdleOrCheckCompatible that is introduced in PIP-44 and #5227.

[gaoran10]

the topic's schema should be changed to the consumer's schema even if it's active or has existed data

I have my doubts about the point "even if it has existed data", maybe this will cause the existed data can't be consumed because the schema changed.

[eolivelli]

LGTM

[BewareMyPower]

@gaoran10 Sorry the PR description is outdated, I'll update it.

[BewareMyPower]

This PR's description has been updated, PTAL again @gaoran10 .

[BewareMyPower]

/pulsarbot run-failure-checks

[BewareMyPower]

/pulsarbot run-failure-checks

[codelipenghui]

@codelipenghui I'm not sure if it's the intended design for producer. Currently handleProducer uses tryAddSchema that is introduced in PIP-43 and #5165, while handleSubscribe uses addSchemaIfIdleOrCheckCompatible that is introduced in PIP-44 and #5227.

@BewareMyPower PIP-43 also need to pass the compatibility check strategy, so we can apply the same rule for handling bytes schema at the producer side?

[BewareMyPower]

@codelipenghui Okay, I'll take a look later.

[Haaroon]

Which version was this fixed in? As of 2.8.1 via following the PublishFunction example, this still raises an error in localrun