From d9e61e64b6dddc793188799d5cb62989b142f862 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 10 Mar 2021 00:52:09 +0800 Subject: [PATCH 1/5] Fix schema not added when subscribe a topic without schema --- .../service/persistent/PersistentTopic.java | 5 +- .../pulsar/client/api/SimpleSchemaTest.java | 51 ++++++++++++++----- 2 files changed, 41 insertions(+), 15 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 25efc0f946465..1376460b37362 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -2510,8 +2510,9 @@ private int getMessageTTL() throws Exception { public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) { return hasSchema() .thenCompose((hasSchema) -> { - if (hasSchema || isActive(InactiveTopicDeleteMode.delete_when_no_subscriptions) - || ledger.getTotalSize() != 0) { + final boolean activeOrNotEmpty = + isActive(InactiveTopicDeleteMode.delete_when_no_subscriptions) || (ledger.getTotalSize() != 0); + if (hasSchema && activeOrNotEmpty) { return checkSchemaCompatibleForConsumer(schema); } else { return addSchema(schema).thenCompose(schemaVersion -> diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java index 6e6e09973d199..b15fe80cf70df 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java @@ -27,6 +27,7 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; import org.apache.avro.reflect.ReflectData; import org.apache.avro.Schema.Parser; @@ -45,6 +46,7 @@ import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaInfo; +import org.apache.pulsar.common.schema.SchemaType; import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; @@ -423,19 +425,6 @@ public void newConsumerWithSchemaOnNewTopic() throws Exception { } } - @Test - public void newConsumerWithSchemaOnExistingTopicWithoutSchema() throws Exception { - String topic = "my-property/my-ns/schema-test"; - - try (Producer p = pulsarClient.newProducer().topic(topic).create(); - Consumer c = pulsarClient.newConsumer(Schema.AVRO(V1Data.class)) - .topic(topic).subscriptionName("sub1").subscribe()) { - Assert.fail("Shouldn't be able to consume with a schema from a topic which has no schema set"); - } catch (PulsarClientException e) { - Assert.assertTrue(e instanceof IncompatibleSchemaException); - } - } - @Test public void newConsumerWithSchemaTopicHasSchema() throws Exception { String topic = "my-property/my-ns/schema-test"; @@ -636,4 +625,40 @@ public void testGetSchemaByVersion() throws PulsarClientException, PulsarAdminEx Assert.assertTrue(binaryLookupService.getSchema(TopicName.get(topic), ByteBuffer.allocate(8).putLong(0).array()).get().isPresent()); Assert.assertTrue(binaryLookupService.getSchema(TopicName.get(topic), ByteBuffer.allocate(8).putLong(1).array()).get().isPresent()); } + + @Test + public void testAutoCreatedSchema() throws Exception { + final String topic1 = "persistent://my-property/my-ns/testAutoCreatedSchema-1"; + final String topic2 = "persistent://my-property/my-ns/testAutoCreatedSchema-2"; + + try (Producer producer = pulsarClient.newProducer(Schema.BYTES).topic(topic1).create()) { + // topic1's schema becomes BYTES now. + } + try { + admin.schemas().getSchemaInfo(topic1); + fail("The schema of topic1 should not exist"); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 404); + } + try (Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topic1).create()) { + // Should pass, STRING schema is compatible with BYTES schema + } + Assert.assertEquals(admin.schemas().getSchemaInfo(topic1).getType(), SchemaType.STRING); + + try (Consumer consumer + = pulsarClient.newConsumer(Schema.BYTES).topic(topic2).subscriptionName("sub").subscribe()) { + // topic2's schema becomes BYTES now. + } + try { + admin.schemas().getSchemaInfo(topic2); + fail("The schema of topic2 should not exist"); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 404); + } + try (Consumer consumer + = pulsarClient.newConsumer(Schema.STRING).topic(topic2).subscriptionName("sub").subscribe()) { + // topic2's schema becomes BYTES now. + } + Assert.assertEquals(admin.schemas().getSchemaInfo(topic2).getType(), SchemaType.STRING); + } } From beb1ac7d0f167d5ef29b7aa088230d327cde182b Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 10 Mar 2021 10:05:18 +0800 Subject: [PATCH 2/5] Check if there's any active consumer instead of if the topic can be deleted --- .../pulsar/broker/service/persistent/PersistentTopic.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 1376460b37362..c08bd7ba21232 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -2510,9 +2510,10 @@ private int getMessageTTL() throws Exception { public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) { return hasSchema() .thenCompose((hasSchema) -> { - final boolean activeOrNotEmpty = - isActive(InactiveTopicDeleteMode.delete_when_no_subscriptions) || (ledger.getTotalSize() != 0); - if (hasSchema && activeOrNotEmpty) { + int numActiveConsumers = subscriptions.values().stream() + .map(subscription -> subscription.getConsumers().size()) + .reduce(0, Integer::sum); + if (hasSchema || (numActiveConsumers != 0) || (ledger.getTotalSize() != 0)) { return checkSchemaCompatibleForConsumer(schema); } else { return addSchema(schema).thenCompose(schemaVersion -> From c992a004dc2a26a72f945fb059d40cdca05783f9 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 10 Mar 2021 10:16:25 +0800 Subject: [PATCH 3/5] Simplify tests --- .../pulsar/client/api/SimpleSchemaTest.java | 22 +++++++------------ 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java index b15fe80cf70df..7f21bc7c6ca15 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java @@ -631,34 +631,28 @@ public void testAutoCreatedSchema() throws Exception { final String topic1 = "persistent://my-property/my-ns/testAutoCreatedSchema-1"; final String topic2 = "persistent://my-property/my-ns/testAutoCreatedSchema-2"; - try (Producer producer = pulsarClient.newProducer(Schema.BYTES).topic(topic1).create()) { - // topic1's schema becomes BYTES now. - } + pulsarClient.newProducer(Schema.BYTES).topic(topic1).create().close(); try { + // BYTES schema is treated as no schema admin.schemas().getSchemaInfo(topic1); fail("The schema of topic1 should not exist"); } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 404); } - try (Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topic1).create()) { - // Should pass, STRING schema is compatible with BYTES schema - } + pulsarClient.newProducer(Schema.STRING).topic(topic1).create().close(); + // topic1's schema becomes STRING now Assert.assertEquals(admin.schemas().getSchemaInfo(topic1).getType(), SchemaType.STRING); - try (Consumer consumer - = pulsarClient.newConsumer(Schema.BYTES).topic(topic2).subscriptionName("sub").subscribe()) { - // topic2's schema becomes BYTES now. - } + pulsarClient.newConsumer(Schema.BYTES).topic(topic2).subscriptionName("sub").subscribe().close(); try { + // BYTES schema is treated as no schema admin.schemas().getSchemaInfo(topic2); fail("The schema of topic2 should not exist"); } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 404); } - try (Consumer consumer - = pulsarClient.newConsumer(Schema.STRING).topic(topic2).subscriptionName("sub").subscribe()) { - // topic2's schema becomes BYTES now. - } + pulsarClient.newConsumer(Schema.STRING).topic(topic2).subscriptionName("sub").subscribe().close(); + // topic2's schema becomes STRING now. Assert.assertEquals(admin.schemas().getSchemaInfo(topic2).getType(), SchemaType.STRING); } } From 9246aa1cb29fd7a492096f7b8df4b32e2cdfb07b Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 10 Mar 2021 11:13:58 +0800 Subject: [PATCH 4/5] Check number of active producers and recover deleted test --- .../broker/service/persistent/PersistentTopic.java | 5 ++++- .../apache/pulsar/client/api/SimpleSchemaTest.java | 14 ++++++++++++++ 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index c08bd7ba21232..325368179e0be 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -2513,7 +2513,10 @@ public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schem int numActiveConsumers = subscriptions.values().stream() .map(subscription -> subscription.getConsumers().size()) .reduce(0, Integer::sum); - if (hasSchema || (numActiveConsumers != 0) || (ledger.getTotalSize() != 0)) { + if (hasSchema + || (!producers.isEmpty()) + || (numActiveConsumers != 0) + || (ledger.getTotalSize() != 0)) { return checkSchemaCompatibleForConsumer(schema); } else { return addSchema(schema).thenCompose(schemaVersion -> diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java index 7f21bc7c6ca15..f4efcffcb2c66 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java @@ -56,6 +56,7 @@ import java.io.ByteArrayInputStream; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.ExecutionException; @@ -425,6 +426,19 @@ public void newConsumerWithSchemaOnNewTopic() throws Exception { } } + @Test + public void newConsumerWithSchemaOnExistingTopicWithoutSchema() throws Exception { + String topic = "my-property/my-ns/schema-test"; + + try (Producer p = pulsarClient.newProducer().topic(topic).create(); + Consumer c = pulsarClient.newConsumer(Schema.AVRO(V1Data.class)) + .topic(topic).subscriptionName("sub1").subscribe()) { + Assert.fail("Shouldn't be able to consume with a schema from a topic which has no schema set"); + } catch (PulsarClientException e) { + Assert.assertTrue(e instanceof IncompatibleSchemaException); + } + } + @Test public void newConsumerWithSchemaTopicHasSchema() throws Exception { String topic = "my-property/my-ns/schema-test"; From 41eb25bd9583bc00cd10b44434ea73563e74845d Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 10 Mar 2021 16:18:27 +0800 Subject: [PATCH 5/5] Fix addSchemaIfIdleOrCheckCompatible for non-persistent topics --- .../service/nonpersistent/NonPersistentTopic.java | 8 +++++++- .../broker/service/persistent/PersistentTopic.java | 4 ++-- .../apache/pulsar/client/api/SimpleSchemaTest.java | 14 +++++++++----- 3 files changed, 18 insertions(+), 8 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java index 0b1f377f6afec..d23779312439f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java @@ -973,7 +973,13 @@ public CompletableFuture getLastMessageId() { @Override public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) { return hasSchema().thenCompose((hasSchema) -> { - if (hasSchema || isActive() || ENTRIES_ADDED_COUNTER_UPDATER.get(this) != 0) { + int numActiveConsumers = subscriptions.values().stream() + .mapToInt(subscription -> subscription.getConsumers().size()) + .sum(); + if (hasSchema + || (!producers.isEmpty()) + || (numActiveConsumers != 0) + || ENTRIES_ADDED_COUNTER_UPDATER.get(this) != 0) { return checkSchemaCompatibleForConsumer(schema); } else { return addSchema(schema).thenCompose(schemaVersion -> CompletableFuture.completedFuture(null)); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 325368179e0be..3ef1b9e945e65 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -2511,8 +2511,8 @@ public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schem return hasSchema() .thenCompose((hasSchema) -> { int numActiveConsumers = subscriptions.values().stream() - .map(subscription -> subscription.getConsumers().size()) - .reduce(0, Integer::sum); + .mapToInt(subscription -> subscription.getConsumers().size()) + .sum(); if (hasSchema || (!producers.isEmpty()) || (numActiveConsumers != 0) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java index f4efcffcb2c66..f670ef19da2ef 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java @@ -56,7 +56,6 @@ import java.io.ByteArrayInputStream; import java.nio.ByteBuffer; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.ExecutionException; @@ -77,6 +76,11 @@ public static Object[][] schemaValidationModes() { return new Object[][] { { true }, { false } }; } + @DataProvider(name = "topicDomain") + public static Object[] topicDomain() { + return new Object[] { "persistent://", "non-persistent://" }; + } + private final boolean schemaValidationEnforced; @Factory(dataProvider = "schemaValidationModes") @@ -640,10 +644,10 @@ public void testGetSchemaByVersion() throws PulsarClientException, PulsarAdminEx Assert.assertTrue(binaryLookupService.getSchema(TopicName.get(topic), ByteBuffer.allocate(8).putLong(1).array()).get().isPresent()); } - @Test - public void testAutoCreatedSchema() throws Exception { - final String topic1 = "persistent://my-property/my-ns/testAutoCreatedSchema-1"; - final String topic2 = "persistent://my-property/my-ns/testAutoCreatedSchema-2"; + @Test(dataProvider = "topicDomain") + public void testAutoCreatedSchema(String domain) throws Exception { + final String topic1 = domain + "my-property/my-ns/testAutoCreatedSchema-1"; + final String topic2 = domain + "my-property/my-ns/testAutoCreatedSchema-2"; pulsarClient.newProducer(Schema.BYTES).topic(topic1).create().close(); try {