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 25efc0f946465..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 @@ -2510,8 +2510,13 @@ 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) { + int numActiveConsumers = subscriptions.values().stream() + .mapToInt(subscription -> subscription.getConsumers().size()) + .sum(); + 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 6e6e09973d199..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 @@ -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; @@ -74,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") @@ -429,7 +436,7 @@ public void newConsumerWithSchemaOnExistingTopicWithoutSchema() throws Exception try (Producer p = pulsarClient.newProducer().topic(topic).create(); Consumer c = pulsarClient.newConsumer(Schema.AVRO(V1Data.class)) - .topic(topic).subscriptionName("sub1").subscribe()) { + .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); @@ -636,4 +643,34 @@ 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(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 { + // 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); + } + pulsarClient.newProducer(Schema.STRING).topic(topic1).create().close(); + // topic1's schema becomes STRING now + Assert.assertEquals(admin.schemas().getSchemaInfo(topic1).getType(), SchemaType.STRING); + + 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); + } + 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); + } }