diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index 21451d6bbac0c..ab565a5d83848 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -571,7 +571,7 @@ public void loadNamespaceTopics(NamespaceBundle bundle) { try { TopicName topicName = TopicName.get(topic); if (bundle.includes(topicName)) { - CompletableFuture future = brokerService.getOrCreateTopic(topic, null); + CompletableFuture future = brokerService.getOrCreateTopic(topic); if (future != null) { persistentTopics.add(future); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 800545cfefc34..8cd0f016309db 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -1225,7 +1225,7 @@ private Topic getTopicReference(TopicName topicName) { } private Topic getOrCreateTopic(TopicName topicName) { - return pulsar().getBrokerService().getOrCreateTopic(topicName.toString(), null).join(); + return pulsar().getBrokerService().getOrCreateTopic(topicName.toString()).join(); } /** diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index 406d3e55c8d46..9a5fc29a8ff35 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -112,7 +112,6 @@ import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.TopicStats; -import org.apache.pulsar.common.schema.SchemaData; import org.apache.pulsar.common.stats.Metrics; import org.apache.pulsar.common.util.FieldParser; import org.apache.pulsar.common.util.FutureUtil; @@ -451,19 +450,14 @@ public void unloadNamespaceBundlesGracefully() { } public CompletableFuture> getTopicIfExists(final String topic) { - return getTopic(topic, false /* createIfMissing */, null /* schemaData */ ); + return getTopic(topic, false /* createIfMissing */); } public CompletableFuture getOrCreateTopic(final String topic) { - return getOrCreateTopic(topic, null); + return getTopic(topic, true /* createIfMissing */).thenApply(Optional::get); } - public CompletableFuture getOrCreateTopic(final String topic, SchemaData schemaData) { - return getTopic(topic, true /* createIfMissing */, schemaData ).thenApply(Optional::get); - } - - private CompletableFuture> getTopic(final String topic, boolean createIfMissing, - SchemaData schemaData) { + private CompletableFuture> getTopic(final String topic, boolean createIfMissing) { try { CompletableFuture> topicFuture = topics.get(topic); if (topicFuture != null) { @@ -477,8 +471,8 @@ private CompletableFuture> getTopic(final String topic, boolean } final boolean isPersistentTopic = TopicName.get(topic).getDomain().equals(TopicDomain.persistent); return topics.computeIfAbsent(topic, (topicName) -> { - return isPersistentTopic ? this.loadOrCreatePersistentTopic(topicName, createIfMissing, schemaData) - : createNonPersistentTopic(topicName, schemaData); + return isPersistentTopic ? this.loadOrCreatePersistentTopic(topicName, createIfMissing) + : createNonPersistentTopic(topicName); }); } catch (IllegalArgumentException e) { log.warn("[{}] Illegalargument exception when loading topic", topic, e); @@ -495,7 +489,7 @@ private CompletableFuture> getTopic(final String topic, boolean } } - private CompletableFuture> createNonPersistentTopic(String topic, SchemaData schemaData) { + private CompletableFuture> createNonPersistentTopic(String topic) { CompletableFuture> topicFuture = new CompletableFuture<>(); if (!pulsar.getConfiguration().isEnableNonPersistentTopics()) { @@ -526,15 +520,7 @@ private CompletableFuture> createNonPersistentTopic(String topic return null; }); - return topicFuture.thenCompose(ot -> { - if (ot.isPresent()) { - // If a schema is provided, add or validate it before the - // topic is "visible" - return ot.get().addSchema(schemaData).thenApply(schemaVersion -> ot); - } else { - return CompletableFuture.completedFuture(ot); - } - }); + return topicFuture; } private static CompletableFuture failedFuture(Throwable t) { @@ -592,7 +578,7 @@ public PulsarClient getReplicationClient(String cluster) { * @throws RuntimeException */ protected CompletableFuture> loadOrCreatePersistentTopic(final String topic, - boolean createIfMissing, SchemaData schemaData) throws RuntimeException { + boolean createIfMissing) throws RuntimeException { checkTopicNsOwnership(topic); final CompletableFuture> topicFuture = new CompletableFuture<>(); @@ -620,15 +606,7 @@ protected CompletableFuture> loadOrCreatePersistentTopic(final S log.debug("topic-loading for {} added into pending queue", topic); } } - return topicFuture.thenCompose(ot -> { - if (ot.isPresent()) { - // If a schema is provided, add or validate it before the - // topic is "visible" - return ot.get().addSchema(schemaData).thenApply(schemaVersion -> ot); - } else { - return CompletableFuture.completedFuture(ot); - } - }); + return topicFuture; } private void createPersistentTopic(final String topic, boolean createIfMissing, CompletableFuture> topicFuture) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 51c3ee01a462c..500ee97427389 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -57,6 +57,7 @@ import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException; import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException; +import org.apache.pulsar.broker.service.schema.IncompatibleSchemaException; import org.apache.pulsar.broker.service.schema.SchemaRegistryService; import org.apache.pulsar.broker.web.RestException; import org.apache.pulsar.client.api.PulsarClientException; @@ -585,20 +586,23 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { } } - service.getOrCreateTopic(topicName.toString(), schema) + service.getOrCreateTopic(topicName.toString()) .thenCompose(topic -> { if (schema != null) { - return topic.isSchemaCompatible(schema).thenCompose(isCompatible -> { - if (isCompatible) { - return topic.subscribe(ServerCnx.this, subscriptionName, consumerId, - subType, priorityLevel, consumerName, isDurable, - startMessageId, metadata, readCompacted, initialPosition); - } else { - return FutureUtil.failedFuture(new BrokerServiceException( - "Trying to subscribe with incompatible schema" - )); - } - }); + return topic.addSchemaIfIdleOrCheckCompatible(schema) + .thenCompose(isCompatible -> { + if (isCompatible) { + return topic.subscribe(ServerCnx.this, subscriptionName, consumerId, + subType, priorityLevel, consumerName, isDurable, + startMessageId, metadata, + readCompacted, initialPosition); + } else { + return FutureUtil.failedFuture( + new BrokerServiceException( + "Trying to subscribe with incompatible schema" + )); + } + }); } else { return topic.subscribe(ServerCnx.this, subscriptionName, consumerId, subType, priorityLevel, consumerName, isDurable, @@ -792,7 +796,7 @@ protected void handleProducer(final CommandProducer cmdProducer) { log.info("[{}][{}] Creating producer. producerId={}", remoteAddress, topicName, producerId); - service.getOrCreateTopic(topicName.toString(), schema).thenAccept((Topic topic) -> { + service.getOrCreateTopic(topicName.toString()).thenAccept((Topic topic) -> { // Before creating producer, check if backlog quota exceeded // on topic if (topic.isBacklogQuotaExceeded(producerName)) { @@ -827,7 +831,16 @@ protected void handleProducer(final CommandProducer cmdProducer) { if (schema != null) { schemaVersionFuture = topic.addSchema(schema); } else { - schemaVersionFuture = CompletableFuture.completedFuture(SchemaVersion.Empty); + schemaVersionFuture = topic.hasSchema().thenCompose((hasSchema) -> { + CompletableFuture result = new CompletableFuture<>(); + if (hasSchema) { + result.completeExceptionally(new IncompatibleSchemaException( + "Producers cannot connect without a schema to topics with a schema")); + } else { + result.complete(SchemaVersion.Empty); + } + return result; + }); } schemaVersionFuture.exceptionally(exception -> { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java index 7488204fcac11..5837898a588da 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java @@ -131,9 +131,28 @@ void updateRates(NamespaceStats nsStats, NamespaceBundleStats currentBundleStats Position getLastMessageId(); + /** + * Whether a topic has had a schema defined for it. + */ + CompletableFuture hasSchema(); + + /** + * Add a schema to the topic. This will fail if the new schema is incompatible with the current + * schema. + */ CompletableFuture addSchema(SchemaData schema); + /** + * Check if schema is compatible with current topic schema. + */ CompletableFuture isSchemaCompatible(SchemaData schema); + /** + * If the topic is idle (no producers, no entries, no subscribers and no existing schema), + * add the passed schema to the topic. Otherwise, check that the passed schema is compatible + * with what the topic already has. + */ + CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema); + CompletableFuture deleteForcefully(); } 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 281d8438db89b..9527191cea307 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 @@ -1002,6 +1002,15 @@ public void markBatchMessagePublished() { private static final Logger log = LoggerFactory.getLogger(NonPersistentTopic.class); + @Override + public CompletableFuture hasSchema() { + String base = TopicName.get(getName()).getPartitionedTopicName(); + String id = TopicName.get(base).getSchemaName(); + return brokerService.pulsar() + .getSchemaRegistryService() + .getSchema(id).thenApply((schema) -> schema != null); + } + @Override public CompletableFuture addSchema(SchemaData schema) { if (schema == null) { @@ -1023,4 +1032,16 @@ public CompletableFuture isSchemaCompatible(SchemaData schema) { .getSchemaRegistryService() .isCompatibleWithLatestVersion(id, schema); } + + @Override + public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) { + return hasSchema() + .thenCompose((hasSchema) -> { + if (hasSchema || isActive() || ENTRIES_ADDED_COUNTER_UPDATER.get(this) != 0) { + return isSchemaCompatible(schema); + } else { + return addSchema(schema).thenApply((ignore) -> true); + } + }); + } } 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 4d289c42605f4..6c12296a6f169 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 @@ -1790,6 +1790,15 @@ public synchronized OffloadProcessStatus offloadStatus() { private static final Logger log = LoggerFactory.getLogger(PersistentTopic.class); + @Override + public CompletableFuture hasSchema() { + String base = TopicName.get(getName()).getPartitionedTopicName(); + String id = TopicName.get(base).getSchemaName(); + return brokerService.pulsar() + .getSchemaRegistryService() + .getSchema(id).thenApply((schema) -> schema != null); + } + @Override public CompletableFuture addSchema(SchemaData schema) { if (schema == null) { @@ -1811,4 +1820,16 @@ public CompletableFuture isSchemaCompatible(SchemaData schema) { .getSchemaRegistryService() .isCompatibleWithLatestVersion(id, schema); } + + @Override + public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) { + return hasSchema() + .thenCompose((hasSchema) -> { + if (hasSchema || isActive() || ledger.getTotalSize() != 0) { + return isSchemaCompatible(schema); + } else { + return addSchema(schema).thenApply((ignore) -> true); + } + }); + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java index 8cecad4b22ee6..aa0a39eba2586 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java @@ -83,23 +83,25 @@ public CompletableFuture getSchema(String schemaId, SchemaVer @Override @NotNull public CompletableFuture putSchemaIfAbsent(String schemaId, SchemaData schema) { - return checkCompatibilityWithLatest(schemaId, schema).thenCompose(isCompatible -> { - if (isCompatible) { - byte[] context = hashFunction.hashBytes(schema.getData()).asBytes(); - SchemaRegistryFormat.SchemaInfo info = SchemaRegistryFormat.SchemaInfo.newBuilder() - .setType(Functions.convertFromDomainType(schema.getType())) - .setSchema(ByteString.copyFrom(schema.getData())) - .setSchemaId(schemaId) - .setUser(schema.getUser()) - .setDeleted(false) - .setTimestamp(clock.millis()) - .addAllProps(toPairs(schema.getProps())) - .build(); - return schemaStorage.put(schemaId, info.toByteArray(), context); - } else { - return FutureUtil.failedFuture(new IncompatibleSchemaException()); - } - }); + return getSchema(schemaId).thenApply( + (existingSchema) -> existingSchema == null || isCompatible(existingSchema, schema)) + .thenCompose(isCompatible -> { + if (isCompatible) { + byte[] context = hashFunction.hashBytes(schema.getData()).asBytes(); + SchemaRegistryFormat.SchemaInfo info = SchemaRegistryFormat.SchemaInfo.newBuilder() + .setType(Functions.convertFromDomainType(schema.getType())) + .setSchema(ByteString.copyFrom(schema.getData())) + .setSchemaId(schemaId) + .setUser(schema.getUser()) + .setDeleted(false) + .setTimestamp(clock.millis()) + .addAllProps(toPairs(schema.getProps())) + .build(); + return schemaStorage.put(schemaId, info.toByteArray(), context); + } else { + return FutureUtil.failedFuture(new IncompatibleSchemaException()); + } + }); } @Override @@ -135,15 +137,14 @@ private SchemaRegistryFormat.SchemaInfo deleted(String schemaId, String user) { .build(); } - private CompletableFuture checkCompatibilityWithLatest(String schemaId, SchemaData schema) { + private boolean isCompatible(SchemaAndMetadata existingSchema, SchemaData newSchema) { + return compatibilityChecks.getOrDefault(newSchema.getType(), SchemaCompatibilityCheck.DEFAULT) + .isCompatible(existingSchema.schema, newSchema); + } - return getSchema(schemaId).thenApply(storedSchema -> - (storedSchema == null) || - compatibilityChecks.getOrDefault( - schema.getType(), - SchemaCompatibilityCheck.DEFAULT - ).isCompatible(storedSchema.schema, schema) - ); + private CompletableFuture checkCompatibilityWithLatest(String schemaId, SchemaData schema) { + return getSchema(schemaId).thenApply( + (existingSchema) -> existingSchema != null && isCompatible(existingSchema, schema)); } interface Functions { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java index 7a068234c661b..94823ad08e55b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java @@ -171,7 +171,7 @@ public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception { consumer.acknowledge(msg); } - PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topic1, null).get(); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topic1).get(); ManagedCursorImpl cursor = (ManagedCursorImpl) topic.getManagedLedger().getCursors().iterator().next(); retryStrategically((test) -> cursor.getState().equals("Open"), 5, 100); @@ -206,7 +206,7 @@ public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception { } // (5) Broker should create new cursor-ledger and remove old cursor-ledger - topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topic1, null).get(); + topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topic1).get(); final ManagedCursorImpl cursor1 = (ManagedCursorImpl) topic.getManagedLedger().getCursors().iterator().next(); retryStrategically((test) -> cursor1.getState().equals("Open"), 5, 100); long newCursorLedgerId = cursor1.getCursorLedger(); @@ -261,7 +261,7 @@ public void testSkipCorruptDataLedger() throws Exception { Consumer consumer = client.newConsumer().topic(topic1).subscriptionName("my-subscriber-name") .receiverQueueSize(5).subscribe(); - PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topic1, null).get(); + PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topic1).get(); ManagedLedgerImpl ml = (ManagedLedgerImpl) topic.getManagedLedger(); ManagedCursorImpl cursor = (ManagedCursorImpl) ml.getCursors().iterator().next(); Field configField = ManagedCursorImpl.class.getDeclaredField("config"); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java index b9b1b847d651e..54d1644975e96 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java @@ -105,7 +105,7 @@ public void testOwnedNsCheck() throws Exception { BrokerService service = pulsar.getBrokerService(); final CountDownLatch latch1 = new CountDownLatch(1); - service.getOrCreateTopic(topic, null).thenAccept(t -> { + service.getOrCreateTopic(topic).thenAccept(t -> { latch1.countDown(); fail("should fail as NS is not owned"); }).exceptionally(exception -> { @@ -118,7 +118,7 @@ public void testOwnedNsCheck() throws Exception { admin.lookups().lookupTopic(topic); final CountDownLatch latch2 = new CountDownLatch(1); - service.getOrCreateTopic(topic, null).thenAccept(t -> { + service.getOrCreateTopic(topic).thenAccept(t -> { try { assertNotNull(service.getTopicReference(topic)); } catch (Exception e) { @@ -747,7 +747,7 @@ public void testTopicLoadingOnDisableNamespaceBundle() throws Exception { // try to create topic which should fail as bundle is disable CompletableFuture> futureResult = pulsar.getBrokerService() - .loadOrCreatePersistentTopic(topicName, true, null); + .loadOrCreatePersistentTopic(topicName, true); try { futureResult.get(); @@ -790,7 +790,7 @@ public void testTopicFailureShouldNotHaveDeadLock() { // create topic async and wait on the future completion executor.submit(() -> { - service.getOrCreateTopic(deadLockTestTopic, null).thenAccept(topic -> topicCreation.complete(null)).exceptionally(e -> { + service.getOrCreateTopic(deadLockTestTopic).thenAccept(topic -> topicCreation.complete(null)).exceptionally(e -> { topicCreation.completeExceptionally(e.getCause()); return null; }); @@ -842,7 +842,7 @@ public void testLedgerOpenFailureShouldNotHaveDeadLock() throws Exception { // create topic async and wait on the future completion executor.submit(() -> { - service.getOrCreateTopic(deadLockTestTopic, null).thenAccept(topic -> topicCreation.complete(null)).exceptionally(e -> { + service.getOrCreateTopic(deadLockTestTopic).thenAccept(topic -> topicCreation.complete(null)).exceptionally(e -> { topicCreation.completeExceptionally(e.getCause()); return null; }); 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 6b01cd6d8364e..e6de6bf62dd64 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 @@ -18,8 +18,10 @@ */ package org.apache.pulsar.client.api; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.testng.Assert.assertEquals; +import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @@ -41,23 +43,147 @@ protected void cleanup() throws Exception { @Test public void testString() throws Exception { - Consumer consumer = pulsarClient.newConsumer(Schema.STRING) - .topic("persistent://my-property/my-ns/my-topic1").subscriptionName("my-subscriber-name").subscribe(); + try (Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic("persistent://my-property/my-ns/my-topic1") + .subscriptionName("my-subscriber-name").subscribe(); + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic("persistent://my-property/my-ns/my-topic1").create()) { + int N = 10; - Producer producer = pulsarClient.newProducer(Schema.STRING) - .topic("persistent://my-property/my-ns/my-topic1").create(); + for (int i = 0; i < N; i++) { + producer.send("my-message-" + i); + } - int N = 10; + for (int i = 0; i < N; i++) { + Message msg = consumer.receive(); + assertEquals(msg.getValue(), "my-message-" + i); - for (int i = 0; i < N; i++) { - producer.send("my-message-" + i); + consumer.acknowledge(msg); + } } + } + + static class V1Data { + int i; + + V1Data() { + this.i = 0; + } + + V1Data(int i) { + this.i = i; + } + + @Override + public int hashCode() { + return i; + } + + @Override + public boolean equals(Object other) { + return (other instanceof V1Data) && i == ((V1Data)other).i; + } + } + + @Test + public void newProducerNewTopicNewSchema() throws Exception { + String topic = "my-property/my-ns/schema-test"; + try (Producer p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)) + .topic(topic).create()) { + p.send(new V1Data(0)); + } + } + + @Test + public void newProducerTopicExistsWithoutSchema() throws Exception { + String topic = "my-property/my-ns/schema-test"; + try (Producer p = pulsarClient.newProducer().topic(topic).create()) { + p.send(topic.getBytes(UTF_8)); + } + + try (Producer p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)) + .topic(topic).create()) { + p.send(new V1Data(0)); + } + } + + @Test + public void newProducerTopicExistsWithSchema() throws Exception { + String topic = "my-property/my-ns/schema-test"; + try (Producer p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)) + .topic(topic).create()) { + p.send(new V1Data(1)); + } + + try (Producer p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)) + .topic(topic).create()) { + p.send(new V1Data(0)); + } + } + + @Test + public void newProducerWithoutSchemaOnTopicWithSchema() throws Exception { + String topic = "my-property/my-ns/schema-test"; - for (int i = 0; i < N; i++) { - Message msg = consumer.receive(); - assertEquals(msg.getValue(), "my-message-" + i); + try (Producer p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)) + .topic(topic).create()) { + p.send(new V1Data(0)); + } + + try (Producer p = pulsarClient.newProducer().topic(topic).create()) { + Assert.fail("Shouldn't be able to connect to a schema'd topic with no schema"); + } catch (PulsarClientException e) { + Assert.assertTrue(e.getMessage().contains("IncompatibleSchemaException")); + } + } + + @Test + public void newConsumerWithSchemaOnNewTopic() throws Exception { + String topic = "my-property/my-ns/schema-test"; + + try (Consumer c = pulsarClient.newConsumer(Schema.AVRO(V1Data.class)) + .topic(topic).subscriptionName("sub1").subscribe(); + Producer p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topic).create()) { + V1Data toSend = new V1Data(1); + p.send(toSend); + Assert.assertEquals(toSend, c.receive().getValue()); + } + } + + @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.getMessage().contains("Trying to subscribe with incompatible schema")); + } + } + + @Test + public void newConsumerWithSchemaTopicHasSchema() throws Exception { + String topic = "my-property/my-ns/schema-test"; + + try (Producer p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topic).create(); + Consumer c = pulsarClient.newConsumer(Schema.AVRO(V1Data.class)) + .topic(topic).subscriptionName("sub1").subscribe()) { + V1Data toSend = new V1Data(1); + p.send(toSend); + Assert.assertEquals(toSend, c.receive().getValue()); + } + } + + @Test + public void newBytesConsumerWithTopicWithSchema() throws Exception { + String topic = "my-property/my-ns/schema-test"; - consumer.acknowledge(msg); + try (Producer p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topic).create(); + Consumer c = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1").subscribe()) { + p.send(new V1Data(1)); + Assert.assertTrue(c.receive().getValue().length > 0); } } }