Skip to content

Clarify and add tests for schema change scenarios #2669

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sijie merged 1 commit into from
Sep 30, 2018
Merged

Clarify and add tests for schema change scenarios #2669

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -571,7 +571,7 @@ public void loadNamespaceTopics(NamespaceBundle bundle) {
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);
}
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}

/**
Expand Down
40 changes: 9 additions & 31 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -451,19 +450,14 @@ public void unloadNamespaceBundlesGracefully() {
}

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) {
Expand All @@ -477,8 +471,8 @@ private CompletableFuture<Optional<Topic>> 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);
Expand All @@ -495,7 +489,7 @@ private CompletableFuture<Optional<Topic>> getTopic(final String topic, boolean
}
}

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()) {
Expand Down Expand Up @@ -526,15 +520,7 @@ private CompletableFuture<Optional<Topic>> 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 <T> CompletableFuture<T> failedFuture(Throwable t) {
Expand Down Expand Up @@ -592,7 +578,7 @@ public PulsarClient getReplicationClient(String cluster) {
* @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<>();
Expand Down Expand Up @@ -620,15 +606,7 @@ protected CompletableFuture<Optional<Topic>> 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<Optional<Topic>> topicFuture) {
Expand Down
41 changes: 27 additions & 14 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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)) {
Expand Down Expand Up @@ -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<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 -> {
Expand Down
19 changes: 19 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,9 +131,28 @@ void updateRates(NamespaceStats nsStats, NamespaceBundleStats currentBundleStats

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();
}
21 changes: 21 additions & 0 deletions ...oker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1002,6 +1002,15 @@ public void markBatchMessagePublished() {

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) {
Expand All @@ -1023,4 +1032,16 @@ public CompletableFuture<Boolean> isSchemaCompatible(SchemaData schema) {
.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);
}
});
}
}
21 changes: 21 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1790,6 +1790,15 @@ public synchronized OffloadProcessStatus offloadStatus() {

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) {
Expand All @@ -1811,4 +1820,16 @@ public CompletableFuture<Boolean> isSchemaCompatible(SchemaData schema) {
.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);
}
});
}
}
51 changes: 26 additions & 25 deletions ...oker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,23 +83,25 @@ public CompletableFuture<SchemaAndMetadata> getSchema(String schemaId, SchemaVer
@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
Expand Down Expand Up @@ -135,15 +137,14 @@ private SchemaRegistryFormat.SchemaInfo deleted(String schemaId, String user) {
.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 {
Expand Down
6 changes: 3 additions & 3 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -261,7 +261,7 @@ public void testSkipCorruptDataLedger() throws Exception {
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");
Expand Down
Loading