fix tests

Author: poorbarcode

Diff for: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java

@@ -29,15 +29,13 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 import javax.ws.rs.core.Response;
-import org.apache.avro.Schema;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.pulsar.broker.admin.AdminResource;
 import org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage;
 import org.apache.pulsar.broker.service.schema.SchemaRegistry.SchemaAndMetadata;
 import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
 import org.apache.pulsar.broker.web.RestException;
 import org.apache.pulsar.client.impl.schema.SchemaUtils;
-import org.apache.pulsar.client.impl.schema.util.SchemaUtil;
 import org.apache.pulsar.client.internal.DefaultImplementation;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
@@ -125,46 +123,8 @@ public CompletableFuture<SchemaVersion> deleteSchemaAsync(boolean authoritative,
                 });
     }
 
-    protected CompletableFuture<Void> checkSchemaTypeSupported(PostSchemaPayload payload) {
-        switch (SchemaType.valueOf(payload.getType())) {
-            case AVRO : {
-                Schema schema = SchemaUtil.parseAvroSchema(payload.getSchema());
-                switch (schema.getType()) {
-                    case RECORD: {
-                        break;
-                    }
-                    case UNION: {
-                        return CompletableFuture.failedFuture(new RestException(
-                                Response.Status.BAD_REQUEST.getStatusCode(),
-                                "[" + String.valueOf(topicName) + "] Avro schema typed [UNION] is not supported"));
-                    }
-                    case INT:
-                    case LONG:
-                    case FLOAT:
-                    case DOUBLE:
-                    case BOOLEAN:
-                    case STRING:
-                    case BYTES: {
-                        return CompletableFuture.failedFuture(new RestException(
-                                Response.Status.BAD_REQUEST.getStatusCode(),
-                                "[" + String.valueOf(topicName) + "] Please call"
-                                + " org.apache.pulsar.client.api.Schema." + schema.getType()
-                                + " when using a simple type schema"));
-                    }
-                    default: {
-                        // ARRAY, MAP, FIXED, NULL.
-                        log.info("[{}] is using a special schema typed [{}]",
-                            String.valueOf(topicName), schema.getType());
-                    }
-                }
-            }
-        }
-        return CompletableFuture.completedFuture(null);
-    }
-
     public CompletableFuture<SchemaVersion> postSchemaAsync(PostSchemaPayload payload, boolean authoritative) {
-        return checkSchemaTypeSupported(payload)
-                .thenCompose(__ -> validateOwnershipAndOperationAsync(authoritative, TopicOperation.PRODUCE))
+        return validateOwnershipAndOperationAsync(authoritative, TopicOperation.PRODUCE)
                 .thenCompose(__ -> getSchemaCompatibilityStrategyAsyncWithoutAuth())
                 .thenCompose(schemaCompatibilityStrategy -> {
                     byte[] data;

Diff for: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidator.java

@@ -22,12 +22,16 @@
 import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException;
 import org.apache.pulsar.common.protocol.schema.SchemaData;
 import org.apache.pulsar.common.schema.KeyValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A validator to validate the schema data is well formed.
  */
 public interface SchemaDataValidator {
 
+    Logger logger = LoggerFactory.getLogger(SchemaDataValidator.class);
+
     /**
      * Validate if the schema data is well formed.
      *

Diff for: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/StructSchemaDataValidator.java

@@ -51,7 +51,10 @@ public void validate(SchemaData schemaData) throws InvalidSchemaDataException {
         try {
             Schema.Parser avroSchemaParser = new Schema.Parser();
             avroSchemaParser.setValidateDefaults(false);
-            avroSchemaParser.parse(new String(data, UTF_8));
+            Schema schema = avroSchemaParser.parse(new String(data, UTF_8));
+            if (SchemaType.AVRO.equals(schemaData.getType())) {
+                checkAvroSchemaTypeSupported(schema);
+            }
         } catch (SchemaParseException e) {
             if (schemaData.getType() == SchemaType.JSON) {
                 // we used JsonSchema for storing the definition of a JSON schema
@@ -70,6 +73,33 @@ public void validate(SchemaData schemaData) throws InvalidSchemaDataException {
         }
     }
 
+    static void checkAvroSchemaTypeSupported(Schema schema) throws InvalidSchemaDataException {
+            switch (schema.getType()) {
+                case RECORD: {
+                    break;
+                }
+                case UNION: {
+                    throw new InvalidSchemaDataException(
+                            "Avro schema typed [UNION] is not supported");
+                }
+                case INT:
+                case LONG:
+                case FLOAT:
+                case DOUBLE:
+                case BOOLEAN:
+                case STRING:
+                case BYTES: {
+                    throw new InvalidSchemaDataException("Please call"
+                            + " org.apache.pulsar.client.api.Schema." + schema.getType()
+                            + " when using a simple type schema");
+                }
+                default: {
+                    // ARRAY, MAP, FIXED, NULL.
+                    logger.info("Registering a special avro schema typed [{}]", schema.getType());
+                }
+            }
+    }
+
     private static void throwInvalidSchemaDataException(SchemaData schemaData,
                                                         Throwable cause) throws InvalidSchemaDataException {
         throw new InvalidSchemaDataException("Invalid schema definition data for "