Return an error if schema is incompatible (apache#1692)

* Return an error is schema is incompatible * Handle string versions correctly * Handle string versions correctly * fix header and update status code for incompatible schemas
yciabaud · May 1, 2018 · 28eb372 · 28eb372
1 parent 762036c
commit 28eb372
Show file tree

Hide file tree

Showing 3 changed files with 48 additions and 7 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/SchemasResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/SchemasResource.java
@@ -18,17 +18,16 @@
  */
 package org.apache.pulsar.broker.admin.v2;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
 import static java.util.Objects.isNull;
 import static org.apache.commons.lang.StringUtils.defaultIfEmpty;
 import static org.apache.pulsar.common.util.Codec.decode;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Charsets;
 import io.swagger.annotations.ApiOperation;
+import java.nio.ByteBuffer;
 import java.time.Clock;
-import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.Encoded;
@@ -42,7 +41,7 @@
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import org.apache.pulsar.broker.admin.AdminResource;
-import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.schema.IncompatibleSchemaException;
 import org.apache.pulsar.broker.web.RestException;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.schema.DeleteSchemaResponse;
@@ -86,6 +85,8 @@ public void getSchema(
                 if (isNull(error)) {
                     if (isNull(schema)) {
                         response.resume(Response.status(Response.Status.NOT_FOUND).build());
+                    } else if (schema.schema.isDeleted()) {
+                        response.resume(Response.noContent());
                     } else {
                         response.resume(
                             Response.ok()
@@ -122,7 +123,9 @@ public void getSchema(
         validateDestinationAndAdminOperation(tenant, namespace, topic);
 
         String schemaId = buildSchemaId(tenant, namespace, topic);
-        SchemaVersion v = pulsar().getSchemaRegistryService().versionFromBytes(version.getBytes());
+        ByteBuffer bbVersion = ByteBuffer.allocate(Long.SIZE);
+        bbVersion.putLong(Long.parseLong(version));
+        SchemaVersion v = pulsar().getSchemaRegistryService().versionFromBytes(bbVersion.array());
         pulsar().getSchemaRegistryService().getSchema(schemaId, v)
             .handle((schema, error) -> {
                 if (isNull(error)) {
@@ -212,7 +215,16 @@ public void postSchema(
                         .build()
                 ).build()
             )
-        );
+        ).exceptionally(error -> {
+            if (error instanceof IncompatibleSchemaException) {
+                response.resume(Response.status(Response.Status.CONFLICT).build());
+            } else {
+                response.resume(
+                    Response.serverError().build()
+                );
+            }
+            return null;
+        });
     }
 
     private String buildSchemaId(String tenant, String namespace, String topic) {

diff --git a/...er/src/main/java/org/apache/pulsar/broker/service/schema/IncompatibleSchemaException.java b/...er/src/main/java/org/apache/pulsar/broker/service/schema/IncompatibleSchemaException.java
@@ -0,0 +1,29 @@
+/**
+ * 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.broker.service.schema;
+
+public class IncompatibleSchemaException extends Exception {
+    public IncompatibleSchemaException() {
+        super("Incompatible schema used");
+    }
+
+    public IncompatibleSchemaException(String message) {
+        super(message);
+    }
+}
diff --git a/...oker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java b/...oker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
@@ -97,7 +97,7 @@ public CompletableFuture<SchemaVersion> putSchemaIfAbsent(String schemaId, Schem
                     .build();
                 return schemaStorage.put(schemaId, info.toByteArray(), context);
             } else {
-                return FutureUtil.failedFuture(new Exception());
+                return FutureUtil.failedFuture(new IncompatibleSchemaException());
             }
         });
     }