You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by iv...@apache.org on 2019/07/08 19:07:26 UTC
[pulsar] branch master updated: Don't consider deleted schema when
checking compatibility (#4669)
This is an automated email from the ASF dual-hosted git repository.
ivank 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 d77980d Don't consider deleted schema when checking compatibility (#4669)
d77980d is described below
commit d77980d2a63236b132cfc6ca7f596de9fb0e7569
Author: Ivan Kelly <iv...@apache.org>
AuthorDate: Mon Jul 8 21:07:20 2019 +0200
Don't consider deleted schema when checking compatibility (#4669)
feca5bb changed topic delete logic to delete the schema when the
topic is deleted (though this only seems to be enabled for idle topic
GC). This exposed a bug in compatibility checking whereby if the a
subscription tries to attach to the topic, even if using the same
schema as had been used previously, a compatibility exception will be
thrown.
This is because the topic still appears to have a schema, even though
there is no actual schema data, just a tombstone. I've changed the logic
to return no schema if the schema read back is a tombstone.
The issue doesn't affect producers because the check was already
correct there.
I've also updated the check for transitive compatibility to remove the
prefix of schemas before the deleted schema. Previously this was
throwing an NPE on the broker as it couldn't decode the deleted
schema.
This issue was discovered by failures in the healthcheck. The check
period (5 minutes) was longer than the GC period (60 seconds). I would
expect it to hit quite often in other scenarios also.
---
.../service/schema/SchemaRegistryServiceImpl.java | 42 +++++++---
.../broker/service/schema/SchemaServiceTest.java | 5 +-
.../pulsar/client/impl/SchemaDeleteTest.java | 97 ++++++++++++++++++++++
3 files changed, 130 insertions(+), 14 deletions(-)
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 57246f0..4014365 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
@@ -68,7 +68,13 @@ public class SchemaRegistryServiceImpl implements SchemaRegistryService {
@Override
@NotNull
public CompletableFuture<SchemaAndMetadata> getSchema(String schemaId) {
- return getSchema(schemaId, SchemaVersion.Latest);
+ return getSchema(schemaId, SchemaVersion.Latest).thenApply((schema) -> {
+ if (schema != null && schema.schema.isDeleted()) {
+ return null;
+ } else {
+ return schema;
+ }
+ });
}
@Override
@@ -184,21 +190,35 @@ public class SchemaRegistryServiceImpl implements SchemaRegistryService {
SchemaCompatibilityStrategy strategy) {
return getSchema(schemaId)
.thenApply(
- (existingSchema) ->
- !(existingSchema == null || existingSchema.schema.isDeleted())
- && isCompatible(existingSchema, schema, strategy));
+ (existingSchema) ->
+ !(existingSchema == null || existingSchema.schema.isDeleted())
+ && isCompatible(existingSchema, schema, strategy));
}
private CompletableFuture<Boolean> checkCompatibilityWithAll(String schemaId, SchemaData schema,
SchemaCompatibilityStrategy strategy) {
return getAllSchemas(schemaId)
- .thenCompose(FutureUtils::collect)
- .thenApply(schemaAndMetadataList -> schemaAndMetadataList
- .stream()
- .map(schemaAndMetadata -> schemaAndMetadata.schema)
- .collect(Collectors.toList()))
- .thenApply(schemas -> compatibilityChecks.getOrDefault(schema.getType(), SchemaCompatibilityCheck.DEFAULT)
- .isCompatible(schemas, schema, strategy));
+ .thenCompose(FutureUtils::collect)
+ .thenApply(list -> {
+ // Trim the prefix of schemas before the latest delete.
+ int lastIndex = list.size() - 1;
+ for (int i = lastIndex; i >= 0; i--) {
+ if (list.get(i).schema.isDeleted()) {
+ if (i == lastIndex) { // if the latest schema is a delete, there's no schemas to compare
+ return Collections.<SchemaAndMetadata>emptyList();
+ } else {
+ return list.subList(i + 1, list.size());
+ }
+ }
+ }
+ return list;
+ })
+ .thenApply(schemaAndMetadataList -> schemaAndMetadataList
+ .stream()
+ .map(schemaAndMetadata -> schemaAndMetadata.schema)
+ .collect(Collectors.toList()))
+ .thenApply(schemas -> compatibilityChecks.getOrDefault(schema.getType(), SchemaCompatibilityCheck.DEFAULT)
+ .isCompatible(schemas, schema, strategy));
}
interface Functions {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java
index 291e64e..ffe7532 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.broker.service.schema;
import static org.testng.AssertJUnit.assertEquals;
import static org.testng.AssertJUnit.assertFalse;
+import static org.testng.AssertJUnit.assertNull;
import static org.testng.AssertJUnit.assertTrue;
import java.time.Clock;
@@ -106,9 +107,7 @@ public class SchemaServiceTest extends MockedPulsarServiceBaseTest {
deleteSchema(schemaId1, version(1));
- SchemaData latest2 = getLatestSchema(schemaId1, version(1));
-
- assertTrue(latest2.isDeleted());
+ assertNull(schemaRegistryService.getSchema(schemaId1).get());
}
@Test
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SchemaDeleteTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SchemaDeleteTest.java
new file mode 100644
index 0000000..24380e1
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SchemaDeleteTest.java
@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import com.google.common.collect.Sets;
+
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
+
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class SchemaDeleteTest extends MockedPulsarServiceBaseTest {
+
+ private static final String subscription = "reader-sub";
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+
+ super.internalSetup();
+ this.conf.setBrokerDeleteInactiveTopicsFrequencySeconds(5);
+
+ admin.clusters().createCluster("test",
+ new ClusterData("http://127.0.0.1:" + BROKER_WEBSERVICE_PORT));
+ admin.tenants().createTenant("my-property",
+ new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
+ admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));
+ }
+
+ @AfterMethod
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void createTopicDeleteTopicCreateTopic() throws Exception {
+ String namespace = "my-property/my-ns";
+ String topic = namespace + "/topic1";
+ String foobar = "foo";
+
+ try (Producer<String> producer =
+ pulsarClient.newProducer(Schema.STRING).topic(topic).create()) {
+ producer.send(foobar);
+ }
+
+ admin.topics().delete(topic);
+ admin.schemas().deleteSchema(topic);
+
+ // creating a subscriber will check the schema against the latest
+ // schema, which in this case should be a tombstone, which should
+ // behave as if the schema never existed
+ try (Reader<String> reader = pulsarClient.newReader(Schema.STRING)
+ .topic(topic).startMessageId(MessageId.latest).create()) {
+ }
+
+ admin.namespaces().setSchemaAutoUpdateCompatibilityStrategy(namespace,
+ SchemaAutoUpdateCompatibilityStrategy.BackwardTransitive);
+ admin.topics().delete(topic);
+ admin.schemas().deleteSchema(topic);
+
+ // with a transitive policy we should check all previous schemas. But we
+ // shouldn't check against those that were there before we deleted the schema.
+ try (Reader<DummyPojo> reader = pulsarClient.newReader(Schema.AVRO(DummyPojo.class))
+ .topic(topic).startMessageId(MessageId.latest).create()) {
+ }
+ }
+
+ public static class DummyPojo {
+ int foobar;
+ }
+}