You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by si...@apache.org on 2018/09/30 18:36:25 UTC
[pulsar] branch master updated: Clarify and add tests for schema
change scenarios (#2669)
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new ff6de4a Clarify and add tests for schema change scenarios (#2669)
ff6de4a is described below
commit ff6de4a4f754ed7d14d64903f3baf05bdd90d23c
Author: Ivan Kelly <iv...@apache.org>
AuthorDate: Sun Sep 30 20:36:21 2018 +0200
Clarify and add tests for schema change scenarios (#2669)
A producer with a schema can attach to a topic if
- the topic does not exist
- the topic exists but currently has no schema
- the topic exists and the schema is compatible with the producer
schema
A consumer with a schema can attach to a topic if
- the topic does not exist (the schema from the consumer is added to
the topic)
- the topic exists and the schema is compatible with the consumer
schema
A producer without a schema can attach if
- the topic does not exist
- the topic exists and has no schema
A consumer without a schema can attach if
- the topic exists, either with or without a schema
---
.../org/apache/pulsar/broker/PulsarService.java | 2 +-
.../broker/admin/impl/PersistentTopicsBase.java | 2 +-
.../pulsar/broker/service/BrokerService.java | 40 ++----
.../apache/pulsar/broker/service/ServerCnx.java | 41 ++++--
.../org/apache/pulsar/broker/service/Topic.java | 19 +++
.../service/nonpersistent/NonPersistentTopic.java | 21 +++
.../broker/service/persistent/PersistentTopic.java | 21 +++
.../service/schema/SchemaRegistryServiceImpl.java | 51 +++----
.../broker/service/BrokerBkEnsemblesTests.java | 6 +-
.../pulsar/broker/service/BrokerServiceTest.java | 10 +-
.../apache/pulsar/client/api/SimpleSchemaTest.java | 148 +++++++++++++++++++--
11 files changed, 270 insertions(+), 91 deletions(-)
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 21451d6..ab565a5 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 class PulsarService implements AutoCloseable {
try {
TopicName topicName = TopicName.get(topic);
if (bundle.includes(topicName)) {
- CompletableFuture<Topic> future = brokerService.getOrCreateTopic(topic, null);
+ CompletableFuture<Topic> 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 800545c..8cd0f01 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 @@ public class PersistentTopicsBase extends AdminResource {
}
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 406d3e5..9a5fc29 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.PersistentOfflineTopicStats;
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 class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
}
public CompletableFuture<Optional<Topic>> getTopicIfExists(final String topic) {
- return getTopic(topic, false /* createIfMissing */, null /* schemaData */ );
+ return getTopic(topic, false /* createIfMissing */);
}
public CompletableFuture<Topic> getOrCreateTopic(final String topic) {
- return getOrCreateTopic(topic, null);
+ return getTopic(topic, true /* createIfMissing */).thenApply(Optional::get);
}
- public CompletableFuture<Topic> getOrCreateTopic(final String topic, SchemaData schemaData) {
- return getTopic(topic, true /* createIfMissing */, schemaData ).thenApply(Optional::get);
- }
-
- private CompletableFuture<Optional<Topic>> getTopic(final String topic, boolean createIfMissing,
- SchemaData schemaData) {
+ private CompletableFuture<Optional<Topic>> getTopic(final String topic, boolean createIfMissing) {
try {
CompletableFuture<Optional<Topic>> topicFuture = topics.get(topic);
if (topicFuture != null) {
@@ -477,8 +471,8 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
}
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 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
}
}
- private CompletableFuture<Optional<Topic>> createNonPersistentTopic(String topic, SchemaData schemaData) {
+ private CompletableFuture<Optional<Topic>> createNonPersistentTopic(String topic) {
CompletableFuture<Optional<Topic>> topicFuture = new CompletableFuture<>();
if (!pulsar.getConfiguration().isEnableNonPersistentTopics()) {
@@ -526,15 +520,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
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 <T> CompletableFuture<T> failedFuture(Throwable t) {
@@ -592,7 +578,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
* @throws RuntimeException
*/
protected CompletableFuture<Optional<Topic>> loadOrCreatePersistentTopic(final String topic,
- boolean createIfMissing, SchemaData schemaData) throws RuntimeException {
+ boolean createIfMissing) throws RuntimeException {
checkTopicNsOwnership(topic);
final CompletableFuture<Optional<Topic>> topicFuture = new CompletableFuture<>();
@@ -620,15 +606,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
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<Optional<Topic>> 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 51c3ee0..500ee97 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.namespace.NamespaceService;
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 @@ public class ServerCnx extends PulsarHandler {
}
}
- 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 @@ public class ServerCnx extends PulsarHandler {
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 @@ public class ServerCnx extends PulsarHandler {
if (schema != null) {
schemaVersionFuture = topic.addSchema(schema);
} else {
- schemaVersionFuture = CompletableFuture.completedFuture(SchemaVersion.Empty);
+ schemaVersionFuture = topic.hasSchema().thenCompose((hasSchema) -> {
+ CompletableFuture<SchemaVersion> 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 7488204..5837898 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 @@ public interface Topic {
Position getLastMessageId();
+ /**
+ * Whether a topic has had a schema defined for it.
+ */
+ CompletableFuture<Boolean> hasSchema();
+
+ /**
+ * Add a schema to the topic. This will fail if the new schema is incompatible with the current
+ * schema.
+ */
CompletableFuture<SchemaVersion> addSchema(SchemaData schema);
+ /**
+ * Check if schema is compatible with current topic schema.
+ */
CompletableFuture<Boolean> 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<Boolean> addSchemaIfIdleOrCheckCompatible(SchemaData schema);
+
CompletableFuture<Void> 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 281d843..9527191 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
@@ -1003,6 +1003,15 @@ public class NonPersistentTopic implements Topic {
private static final Logger log = LoggerFactory.getLogger(NonPersistentTopic.class);
@Override
+ public CompletableFuture<Boolean> 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<SchemaVersion> addSchema(SchemaData schema) {
if (schema == null) {
return CompletableFuture.completedFuture(SchemaVersion.Empty);
@@ -1023,4 +1032,16 @@ public class NonPersistentTopic implements Topic {
.getSchemaRegistryService()
.isCompatibleWithLatestVersion(id, schema);
}
+
+ @Override
+ public CompletableFuture<Boolean> 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 4d289c4..6c12296 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
@@ -1791,6 +1791,15 @@ public class PersistentTopic implements Topic, AddEntryCallback {
private static final Logger log = LoggerFactory.getLogger(PersistentTopic.class);
@Override
+ public CompletableFuture<Boolean> 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<SchemaVersion> addSchema(SchemaData schema) {
if (schema == null) {
return CompletableFuture.completedFuture(SchemaVersion.Empty);
@@ -1811,4 +1820,16 @@ public class PersistentTopic implements Topic, AddEntryCallback {
.getSchemaRegistryService()
.isCompatibleWithLatestVersion(id, schema);
}
+
+ @Override
+ public CompletableFuture<Boolean> 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 8cecad4..aa0a39e 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 class SchemaRegistryServiceImpl implements SchemaRegistryService {
@Override
@NotNull
public CompletableFuture<SchemaVersion> 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 @@ public class SchemaRegistryServiceImpl implements SchemaRegistryService {
.build();
}
- private CompletableFuture<Boolean> 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<Boolean> 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 7a06823..94823ad 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 class BrokerBkEnsemblesTests {
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 class BrokerBkEnsemblesTests {
}
// (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 class BrokerBkEnsemblesTests {
Consumer<byte[]> 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 b9b1b84..54d1644 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 class BrokerServiceTest extends BrokerTestBase {
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 class BrokerServiceTest extends BrokerTestBase {
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 class BrokerServiceTest extends BrokerTestBase {
// try to create topic which should fail as bundle is disable
CompletableFuture<Optional<Topic>> futureResult = pulsar.getBrokerService()
- .loadOrCreatePersistentTopic(topicName, true, null);
+ .loadOrCreatePersistentTopic(topicName, true);
try {
futureResult.get();
@@ -790,7 +790,7 @@ public class BrokerServiceTest extends BrokerTestBase {
// 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 class BrokerServiceTest extends BrokerTestBase {
// 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 6b01cd6..e6de6bf 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 @@ public class SimpleSchemaTest extends ProducerConsumerBase {
@Test
public void testString() throws Exception {
- Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
- .topic("persistent://my-property/my-ns/my-topic1").subscriptionName("my-subscriber-name").subscribe();
+ try (Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic1")
+ .subscriptionName("my-subscriber-name").subscribe();
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic1").create()) {
+ int N = 10;
- Producer<String> 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<String> 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<V1Data> 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<byte[]> p = pulsarClient.newProducer().topic(topic).create()) {
+ p.send(topic.getBytes(UTF_8));
+ }
+
+ try (Producer<V1Data> 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<V1Data> p = pulsarClient.newProducer(Schema.AVRO(V1Data.class))
+ .topic(topic).create()) {
+ p.send(new V1Data(1));
+ }
+
+ try (Producer<V1Data> 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<String> msg = consumer.receive();
- assertEquals(msg.getValue(), "my-message-" + i);
+ try (Producer<V1Data> p = pulsarClient.newProducer(Schema.AVRO(V1Data.class))
+ .topic(topic).create()) {
+ p.send(new V1Data(0));
+ }
+
+ try (Producer<byte[]> 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<V1Data> c = pulsarClient.newConsumer(Schema.AVRO(V1Data.class))
+ .topic(topic).subscriptionName("sub1").subscribe();
+ Producer<V1Data> 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<byte[]> p = pulsarClient.newProducer().topic(topic).create();
+ Consumer<V1Data> 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<V1Data> p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topic).create();
+ Consumer<V1Data> 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<V1Data> p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topic).create();
+ Consumer<byte[]> c = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1").subscribe()) {
+ p.send(new V1Data(1));
+ Assert.assertTrue(c.receive().getValue().length > 0);
}
}
}