package io.apicurio.registry.noprofile.ccompat.rest.v7;

import io.apicurio.registry.AbstractResourceTestBase;
import io.apicurio.registry.ccompat.dto.SchemaContent;
import io.apicurio.registry.ccompat.rest.error.ErrorCode;
import io.apicurio.registry.model.GroupId;
import io.apicurio.registry.rest.Headers;
import io.apicurio.registry.rest.client.models.CreateRule;
import io.apicurio.registry.rest.client.models.RuleType;
import io.apicurio.registry.support.HealthUtils;
import io.apicurio.registry.support.TestCmmn;
import io.apicurio.registry.utils.tests.DeletionEnabledProfile;
import io.apicurio.registry.utils.tests.TestUtils;
import io.confluent.connect.avro.AvroConverter;
import io.confluent.kafka.schemaregistry.CompatibilityLevel;
import io.confluent.kafka.schemaregistry.ParsedSchema;
import io.confluent.kafka.schemaregistry.SchemaProvider;
import io.confluent.kafka.schemaregistry.avro.AvroSchema;
import io.confluent.kafka.schemaregistry.avro.AvroSchemaProvider;
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.rest.RestService;
import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference;
import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaString;
import io.confluent.kafka.schemaregistry.client.rest.entities.SubjectVersion;
import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaRequest;
import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
import io.confluent.kafka.schemaregistry.json.JsonSchemaProvider;
import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer;
import io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer;
import io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializerConfig;
import io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer;
import io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializerConfig;
import io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer;
import io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializerConfig;
import io.quarkus.test.junit.QuarkusTest;
import io.quarkus.test.junit.TestProfile;
import org.apache.avro.AvroTypeException;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.kafka.connect.data.SchemaAndValue;
import org.apache.kafka.connect.data.Struct;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import static io.apicurio.registry.utils.tests.TestUtils.retry;
import static io.confluent.kafka.schemaregistry.CompatibilityLevel.BACKWARD;
import static io.confluent.kafka.schemaregistry.CompatibilityLevel.FORWARD;
import static io.confluent.kafka.schemaregistry.CompatibilityLevel.FULL;
import static io.confluent.kafka.schemaregistry.CompatibilityLevel.NONE;
import static java.net.HttpURLConnection.HTTP_CONFLICT;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

@QuarkusTest
@TestProfile(DeletionEnabledProfile.class)
@SuppressWarnings({ "unchecked", "rawtypes" })
public class ConfluentClientTest extends AbstractResourceTestBase {

    public SchemaRegistryClient buildClient() {
        final List<SchemaProvider> schemaProviders = Arrays.asList(new JsonSchemaProvider(),
                new AvroSchemaProvider(), new ProtobufSchemaProvider());
        return new CachedSchemaRegistryClient(
                new RestService("http://localhost:" + testPort + "/apis/ccompat/v7"), 3, schemaProviders,
                null, Map.of(Headers.GROUP_ID, "confluentV7-test-group"));
    }

    @AfterEach
    protected void afterEach() throws Exception {
        try {
            clientV3.groups().byGroupId(GroupId.DEFAULT.getRawGroupIdWithDefaultString()).artifacts()
                    .delete();
        } catch (Exception ignored) {
        }
    }

    @Test
    public void testSerdeProtobufSchema() {
        TestCmmn.UUID record = TestCmmn.UUID.newBuilder().setLsb(2).setMsb(1).build();

        final SchemaRegistryClient client = buildClient();
        final String subject = generateArtifactId();

        final Map<String, Object> config = new HashMap<>();
        config.put(KafkaProtobufSerializerConfig.AUTO_REGISTER_SCHEMAS, true);
        config.put(KafkaProtobufSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG,
                "http://localhost:" + testPort + "/apis/ccompat/v7");
        config.put(KafkaProtobufDeserializerConfig.SPECIFIC_PROTOBUF_VALUE_TYPE,
                TestCmmn.UUID.class.getName());

        try (KafkaProtobufSerializer<TestCmmn.UUID> serializer = new KafkaProtobufSerializer<>(client);
            KafkaProtobufDeserializer<TestCmmn.UUID> deserializer = new KafkaProtobufDeserializer<>(client)) {

            serializer.configure(config, false);
            deserializer.configure(config, false);

            byte[] bytes = serializer.serialize(subject, record);
            TestCmmn.UUID deserialized = deserializer.deserialize(subject, bytes);
            Assertions.assertEquals(record, deserialized);
        }
    }

    @Test
    public void testSmoke() throws Exception {
        SchemaRegistryClient client = buildClient();
        Collection<String> subjects = client.getAllSubjects();
        Assertions.assertNotNull(subjects);
    }

    @Test
    public void testOrphanedContent() throws Exception {
        SchemaRegistryClient client = buildClient();
        final String subject = generateArtifactId();

        String orphanedContentSchema = "{\"type\":\"record\",\"name\":\"testOrphanedContent\",\"fields\":[{\"name\":\"f1\",\"type\":\"string\"}]}";

        // Create schema with the first id
        ParsedSchema schema = new AvroSchema(orphanedContentSchema);
        int id1 = client.register(subject, schema);

        // Reset the client cache so that the next line actually does what we want.
        client.reset();

        // The schema can be fetched with this particular id
        TestUtils.retry(() -> client.getSchemaById(id1));

        // First sotft delete subject, then hard delete, the content must be claimed as orphaned
        client.deleteSubject(subject);
        client.deleteSubject(subject, true);

        // Register schema again, the id must be different
        int id2 = client.register(subject, schema);

        Assertions.assertNotEquals(id1, id2);

        // The schema must be retrievable using the new id
        TestUtils.retry(() -> client.getSchemaById(id2));
    }

    @Test
    public void testSimpleOps() throws Exception {
        SchemaRegistryClient client = buildClient();
        final String subject = generateArtifactId();

        ParsedSchema schema1 = new AvroSchema(
                "{\"type\":\"record\",\"name\":\"myrecord1\",\"fields\":[{\"name\":\"f1\",\"type\":\"string\"}]}");
        int id1 = client.register(subject, schema1);

        // Reset the client cache so that the next line actually does what we want.
        client.reset();

        TestUtils.retry(() -> client.getSchemaById(id1));

        ParsedSchema schema2 = new AvroSchema(
                "{\"type\":\"record\",\"name\":\"myrecord2\",\"fields\":[{\"name\":\"f2\",\"type\":\"string\"}]}");
        int id2 = client.register(subject, schema2);

        TestUtils.retry(() -> client.getSchemaById(id2));

        ParsedSchema schema = client.getSchemaById(id1);
        Assertions.assertNotNull(schema);

        client.reset();

        Assertions.assertTrue(client.testCompatibility(subject, schema2));

        // global id can be mapped async
        retry(() -> {
            ParsedSchema schema3 = client.getSchemaById(id2);
            Assertions.assertNotNull(schema3);
            return schema3;
        });

        Collection<String> subjects = client.getAllSubjects();
        Assertions.assertTrue(subjects.contains(subject));

        List<Integer> versions = client.getAllVersions(subject);
        Assertions.assertTrue(versions.contains(1));
        Assertions.assertTrue(versions.contains(2));

        int v2 = client.getVersion(subject, schema2);
        Assertions.assertEquals(2, v2);

        int d1 = client.deleteSchemaVersion(subject, "1");
        Assertions.assertEquals(1, d1);
        int d2 = client.deleteSchemaVersion(subject, "2");
        Assertions.assertEquals(2, d2);
    }

    @Test
    public void testSerdeAvro() throws Exception {
        SchemaRegistryClient client = buildClient();

        String subject = generateArtifactId();

        String rawSchema = "{\"type\":\"record\",\"name\":\"myrecord3\",\"fields\":[{\"name\":\"bar\",\"type\":\"string\"}]}";
        ParsedSchema schema = new AvroSchema(rawSchema);
        int id = client.register(subject + "-value", schema);
        client.reset();

        // global id can be mapped async
        retry(() -> {
            ParsedSchema schema2 = client.getSchemaById(id);
            Assertions.assertNotNull(schema2);
            return schema2;
        });

        try (KafkaAvroSerializer serializer = new KafkaAvroSerializer(client);
            KafkaAvroDeserializer deserializer = new KafkaAvroDeserializer(client)) {

            GenericData.Record record = new GenericData.Record(new Schema.Parser().parse(rawSchema));
            record.put("bar", "somebar");

            byte[] bytes = serializer.serialize(subject, record);
            GenericData.Record ir = (GenericData.Record) deserializer.deserialize(subject, bytes);

            Assertions.assertEquals("somebar", ir.get("bar").toString());
        }
    }

    @Test
    public void testSerdeJsonSchema() {

        final SchemaRegistryClient client = buildClient();
        final String subject = generateArtifactId();

        final SchemaContent schemaContent = new SchemaContent(
                "{\"type\":\"record\",\"name\":\"myrecord1\",\"fields\":[{\"name\":\"f1\",\"type\":\"string\"}]}");

        final Properties config = new Properties();
        config.put(KafkaJsonSchemaSerializerConfig.AUTO_REGISTER_SCHEMAS, true);
        config.put(KafkaJsonSchemaSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG,
                "http://localhost:" + testPort + "/apis/ccompat/v7");

        try (
            KafkaJsonSchemaSerializer serializer = new KafkaJsonSchemaSerializer(client, new HashMap(config));
            KafkaJsonSchemaDeserializer deserializer = new KafkaJsonSchemaDeserializer(client, config,
                    SchemaContent.class)) {

            byte[] bytes = serializer.serialize(subject, schemaContent);
            Object deserialized = deserializer.deserialize(subject, bytes);
            Assertions.assertEquals(schemaContent, deserialized);
        }
    }

    @Test
    public void testDelete() throws Exception {
        SchemaRegistryClient client = buildClient();

        String nonExisting = generateArtifactId();
        try {
            client.deleteSubject(nonExisting);
            Assertions.fail();
        } catch (RestClientException e) {
            Assertions.assertEquals(404, e.getStatus());
        }

        retry(() -> {
            HealthUtils.assertIsReady(testPort);
            HealthUtils.assertIsLive(testPort);
        });

        String subject = generateArtifactId();

        ParsedSchema schema = new AvroSchema(
                "{\"type\":\"record\",\"name\":\"myrecord3\",\"fields\":[{\"name\":\"bar\",\"type\":\"string\"}]}");
        int id = client.register(subject, schema);
        client.reset();

        // global id can be mapped async
        retry(() -> {
            ParsedSchema schema2 = client.getSchemaById(id);
            Assertions.assertNotNull(schema2);
            return schema2;
        });

        Collection<String> subjects = client.getAllSubjects();
        Assertions.assertTrue(subjects.contains(subject));

        client.deleteSubject(subject);

        retry(() -> {
            // delete can be async
            Collection<String> all = client.getAllSubjects();
            Assertions.assertFalse(all.contains(subject));
            return null;
        });
    }

    /**
     * Test for issue: <a href="https://github.com/Apicurio/apicurio-registry/issues/536">...</a>
     */
    @Test
    public void testGlobalRule() throws Exception {
        SchemaRegistryClient client = buildClient();

        CreateRule createRule = new CreateRule();
        createRule.setRuleType(RuleType.COMPATIBILITY);
        createRule.setConfig("BACKWARD");
        clientV3.admin().rules().post(createRule);

        String subject = generateArtifactId();
        ParsedSchema schema = new AvroSchema(
                "{\"type\":\"record\",\"name\":\"myrecord3\",\"fields\":[{\"name\":\"bar\",\"type\":\"string\"}]}");
        int id = client.register(subject, schema);
        client.reset();

        // global id can be mapped async
        retry(() -> {
            ParsedSchema schema2 = client.getSchemaById(id);
            Assertions.assertNotNull(schema2);
            return schema2;
        });

        // try to register an incompatible schema
        Assertions.assertThrows(RestClientException.class, () -> {
            ParsedSchema schema2 = new AvroSchema("{\"type\":\"string\"}");
            client.register(subject, schema2);
            client.reset();
        });
    }

    @Test
    public void testConverter_PreRegisterSchema() {
        String subject = generateArtifactId();
        String name = "myr" + ThreadLocalRandom.current().nextInt(0, Integer.MAX_VALUE);
        testConverter(subject, name, false, (client) -> {
            try {
                ParsedSchema schema = new AvroSchema(String.format(
                        "{\"type\":\"record\",\"name\":\"%s\",\"fields\":[{\"name\":\"bar\",\"type\":\"string\"}],\"connect.name\":\"%s\"}",
                        name, name));
                int id = client.register(subject + "-value", schema);
                client.reset();
                // can be async ...
                ParsedSchema retry = retry(() -> client.getSchemaById(id));
                Assertions.assertNotNull(retry);
            } catch (Exception e) {
                throw new IllegalStateException(e);
            }
        }, (c, b) -> {
        });
    }

    @Test
    public void testConverter_AutoRegisterSchema() {
        String name = "myr" + ThreadLocalRandom.current().nextInt(0, Integer.MAX_VALUE);
        testConverter(generateArtifactId(), name, true, (c) -> {
        }, (client, bytes) -> {
            try {
                client.reset();
                ParsedSchema retry = retry(() -> {
                    ByteBuffer buffer = ByteBuffer.wrap(bytes);
                    buffer.get(); // magic-byte
                    int id = buffer.getInt();
                    return client.getSchemaById(id);
                });
                Assertions.assertNotNull(retry);
            } catch (Exception e) {
                throw new IllegalStateException(e);
            }
        });
    }

    private void testConverter(String subject, String name, boolean autoRegister,
            Consumer<SchemaRegistryClient> pre, BiConsumer<SchemaRegistryClient, byte[]> post) {
        SchemaRegistryClient client = buildClient();

        pre.accept(client);

        org.apache.kafka.connect.data.Schema cs = org.apache.kafka.connect.data.SchemaBuilder.struct()
                .name(name).field("bar", org.apache.kafka.connect.data.Schema.STRING_SCHEMA);
        Struct struct = new Struct(cs);
        struct.put("bar", "somebar");

        AvroConverter converter = new AvroConverter(client);
        Map<String, Object> config = new HashMap<>();
        config.put("schema.registry.url", "dummy");
        config.put("auto.register.schemas", autoRegister);
        converter.configure(config, false);

        byte[] bytes = converter.fromConnectData(subject, cs, struct);

        post.accept(client, bytes);

        SchemaAndValue sav = converter.toConnectData(subject, bytes);
        Struct ir = (Struct) sav.value();
        Assertions.assertEquals("somebar", ir.get("bar").toString());
    }

    @Test
    public void testSimpleAvroSchema() throws Exception {
        SchemaRegistryClient client = buildClient();
        final String subject = generateArtifactId();

        ParsedSchema schema1 = new AvroSchema("\"string\"");
        int id1 = client.register(subject, schema1);

        // Reset the client cache so that the next line actually does what we want.
        client.reset();

        TestUtils.retry(() -> client.getSchemaById(id1));

        ParsedSchema schema2 = new AvroSchema("{\"type\":\"string\"}");
        int id2 = client.register(subject, schema2);

        TestUtils.retry(() -> client.getSchemaById(id2));
    }

    @Test
    public void testCreateRuleBeforeSchema() throws Exception {
        SchemaRegistryClient client = buildClient();
        final String subject = generateArtifactId();

        client.updateCompatibility(subject, "FULL");

        ParsedSchema schema1 = new AvroSchema("\"string\"");
        int id1 = client.register(subject, schema1);

        // Reset the client cache so that the next line actually does what we want.
        client.reset();

        TestUtils.retry(() -> client.getSchemaById(id1));

        ParsedSchema schema2 = new AvroSchema("{\"type\":\"string\"}");
        int id2 = client.register(subject, schema2);

        TestUtils.retry(() -> client.getSchemaById(id2));
    }

    @Test
    public void testBasicAvro() throws Exception {
        String subject1 = "testBasic1";
        String subject2 = "testBasic2";
        int schemasInSubject1 = 10;
        List<Integer> allVersionsInSubject1 = new ArrayList<>();
        List<String> allSchemasInSubject1 = ConfluentTestUtils
                .getRandomCanonicalAvroString(schemasInSubject1);
        int schemasInSubject2 = 5;
        List<Integer> allVersionsInSubject2 = new ArrayList<>();
        List<String> allSchemasInSubject2 = ConfluentTestUtils
                .getRandomCanonicalAvroString(schemasInSubject2);
        List<String> allSubjects = new ArrayList<>();

        List<Integer> schemaIds = new ArrayList<>();

        // test getAllVersions with no existing data
        try {
            confluentClient.getAllVersions(subject1);
            fail("Getting all versions from non-existing subject1 should fail with "
                    + ErrorCode.SUBJECT_NOT_FOUND.value() + " (subject not found)");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(),
                    "Should get a 404 status for non-existing subject");
        }

        // test getAllSubjects with no existing data
        assertEquals(allSubjects, confluentClient.getAllSubjects(),
                "Getting all subjects should return empty");

        // test registering and verifying new schemas in subject1
        for (int i = 0; i < schemasInSubject1; i++) {
            String schema = allSchemasInSubject1.get(i);
            int expectedVersion = i + 1;
            int schemaId = ConfluentTestUtils.registerAndVerifySchema(confluentClient, schema, subject1);
            schemaIds.add(schemaId);
            allVersionsInSubject1.add(expectedVersion);
        }

        allSubjects.add(subject1);

        // test re-registering existing schemas
        for (int i = 0; i < schemasInSubject1; i++) {
            String schemaString = allSchemasInSubject1.get(i);
            int foundId = confluentClient.registerSchema(schemaString, subject1, true).getId();
            assertEquals((int) schemaIds.get(i), foundId,
                    "Re-registering an existing schema should return the existing version");
        }

        // test registering schemas in subject2
        for (int i = 0; i < schemasInSubject2; i++) {
            String schema = allSchemasInSubject2.get(i);
            int expectedVersion = i + 1;
            ConfluentTestUtils.registerAndVerifySchema(confluentClient, schema, subject2);
            allVersionsInSubject2.add(expectedVersion);
        }
        allSubjects.add(subject2);

        // test getAllVersions with existing data
        assertEquals(allVersionsInSubject1, confluentClient.getAllVersions(subject1),
                "Getting all versions from subject1 should match all registered versions");
        assertEquals(allVersionsInSubject2, confluentClient.getAllVersions(subject2),
                "Getting all versions from subject2 should match all registered versions");

        // test getAllSubjects with existing data
        assertEquals(allSubjects, confluentClient.getAllSubjects(),
                "Getting all subjects should match all registered subjects");
    }

    @Test
    public void testRegisterSameSchemaOnDifferentSubject() throws Exception {
        String schema = ConfluentTestUtils.getRandomCanonicalAvroString(1).get(0);
        int id1 = confluentClient.registerSchema(schema, "subject1", true).getId();
        int id2 = confluentClient.registerSchema(schema, "subject2", true).getId();
        assertEquals(id1, id2,
                "Registering the same schema under different subjects should return the same id");
    }

    @Test
    public void testRegisterInvalidSchemaBadType() throws Exception {
        String subject = "testRegisterInvalidSchemaBadType";

        // Invalid Field Type 'str'
        String badSchemaString = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"str\",\"name\":\"field1\"}]}";

        try {
            new org.apache.avro.Schema.Parser().parse(badSchemaString);
            fail("Parsing invalid schema string should fail with AvroTypeException");
        } catch (AvroTypeException ignored) {
        }

        try {
            confluentClient.registerSchema(badSchemaString, subject, true);
            fail("Registering schema with invalid field type should fail with "
                    + ErrorCode.INVALID_SCHEMA.value() + " (invalid schema)");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.INVALID_SCHEMA.value(), rce.getErrorCode(), "Invalid schema");
        }
    }

    @Test
    public void testRegisterInvalidSchemaBadReference() throws Exception {
        String subject = "testRegisterInvalidSchemaBadReference";

        // Invalid Reference
        SchemaReference invalidReference = new SchemaReference("invalid.schema", "badSubject", 1);
        String schemaString = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":\"field1\"}]}";

        try {
            confluentClient.registerSchema(schemaString, "AVRO", Collections.singletonList(invalidReference),
                    subject, true);
            fail("Registering schema with invalid reference should fail with "
                    + ErrorCode.INVALID_SCHEMA.value() + " (invalid schema)");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.INVALID_SCHEMA.value(), rce.getErrorCode(), "Invalid schema");
        }
    }

    @Test
    public void testCompatibleSchemaLookupBySubject() throws Exception {
        String subject = "testSubject";
        int numRegisteredSchemas = 0;
        int numSchemas = 10;

        List<String> allSchemas = ConfluentTestUtils.getRandomCanonicalAvroString(numSchemas);

        confluentClient.registerSchema(allSchemas.get(0), subject);
        numRegisteredSchemas++;

        // test compatibility of this schema against the latest version under the subject
        String schema1 = allSchemas.get(0);
        boolean isCompatible = confluentClient.testCompatibility(schema1, subject, "latest").isEmpty();
        assertTrue(isCompatible, "First schema registered should be compatible");

        for (int i = 0; i < numSchemas; i++) {
            // Test that compatibility check doesn't change the number of versions
            String schema = allSchemas.get(i);
            assertTrue(confluentClient.testCompatibility(schema, subject, "latest").isEmpty());
            ConfluentTestUtils.checkNumberOfVersions(confluentClient, numRegisteredSchemas, subject);
        }
    }

    @Test
    public void testIncompatibleSchemaLookupBySubject() throws Exception {
        String subject = "testSubject";

        // Make two incompatible schemas - field 'f' has different types
        String schema1String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":" + "\"f" + "\"}]}";
        String schema1 = new AvroSchema(schema1String).canonicalString();

        String schema2String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"int\",\"name\":" + "\"f" + "\"}]}";
        String schema2 = new AvroSchema(schema2String).canonicalString();

        // ensure registering incompatible schemas will raise an error
        confluentClient.updateCompatibility(CompatibilityLevel.FULL.name, subject);

        // test that compatibility check for incompatible schema returns false and the appropriate
        // error response from Avro
        confluentClient.registerSchema(schema1, subject);
        int versionOfRegisteredSchema = confluentClient.lookUpSubjectVersion(schema1, subject).getVersion();
        boolean isCompatible = confluentClient
                .testCompatibility(schema2, subject, String.valueOf(versionOfRegisteredSchema)).isEmpty();
        assertFalse(isCompatible, "Schema should be incompatible with specified version");
    }

    @Test
    public void testIncompatibleSchemaBySubject() throws Exception {
        String subject = "testSubject";

        String schema1String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":\"f1\"},{\"type\":\"string\",\"name\":\"f2\"}]}";
        String schema1 = new AvroSchema(schema1String).canonicalString();

        String schema2String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":\"f1\"}]}";
        String schema2 = new AvroSchema(schema2String).canonicalString();

        String schema3String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":\"f1\"},{\"type\":\"string\",\"name\":\"f3\"}]}";
        String schema3 = new AvroSchema(schema3String).canonicalString();

        confluentClient.registerSchema(schema1, subject);
        confluentClient.registerSchema(schema2, subject);

        confluentClient.updateCompatibility(CompatibilityLevel.FORWARD_TRANSITIVE.name, subject);

        // schema3 is compatible with schema2, but not compatible with schema1
        boolean isCompatible = confluentClient.testCompatibility(schema3, subject, "latest").isEmpty();
        assertTrue(isCompatible, "Schema is compatible with the latest version");
        isCompatible = confluentClient.testCompatibility(schema3, subject, null).isEmpty();
        assertFalse(isCompatible, "Schema should be incompatible with FORWARD_TRANSITIVE setting");
        try {
            confluentClient.registerSchema(schema3String, subject);
            fail("Schema register should fail since schema is incompatible");
        } catch (RestClientException e) {
            assertEquals(HTTP_CONFLICT, e.getErrorCode(),
                    "Schema register should fail since schema is incompatible");
            assertFalse(e.getMessage().isEmpty());
        }
    }

    @Test
    public void testSchemaRegistrationUnderDiffSubjects() throws Exception {
        String subject1 = "testSchemaRegistrationUnderDiffSubjects1";
        String subject2 = "testSchemaRegistrationUnderDiffSubjects2";

        // Make two incompatible schemas - field 'f' has different types
        String schemaString1 = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":" + "\"f" + "\"}]}";
        String schema1 = new AvroSchema(schemaString1).canonicalString();

        String schemaString2 = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"int\",\"name\":" + "\"foo" + "\"}]}";
        String schema2 = new AvroSchema(schemaString2).canonicalString();

        confluentClient.registerSchema(schema1, subject1);
        int versionOfRegisteredSchema1Subject1 = confluentClient.lookUpSubjectVersion(schema1, subject1)
                .getVersion();
        assertEquals(1, versionOfRegisteredSchema1Subject1,
                "1st schema under subject1 should have version 1");

        int idOfRegisteredSchema2Subject1 = confluentClient.registerSchema(schema2, subject1);
        int versionOfRegisteredSchema2Subject1 = confluentClient.lookUpSubjectVersion(schema2, subject1)
                .getVersion();
        assertEquals(2, versionOfRegisteredSchema2Subject1,
                "2nd schema under subject1 should have version 2");

        int idOfRegisteredSchema2Subject2 = confluentClient.registerSchema(schema2, subject2);
        assertEquals(idOfRegisteredSchema2Subject1, idOfRegisteredSchema2Subject2,
                "Since schema is globally registered but not under subject2, id should not change");
    }

    @Test
    public void testConfigDefaults() throws Exception {
        assertEquals(NONE.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "Default compatibility level should be none for this test instance");

        // change it to forward
        confluentClient.updateCompatibility(CompatibilityLevel.FORWARD.name, null);

        assertEquals(FORWARD.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "New compatibility level should be forward for this test instance");
    }

    @Test
    public void testNonExistentSubjectConfigChange() throws Exception {
        String subject = "testSubject";
        try {
            confluentClient.updateCompatibility(CompatibilityLevel.FORWARD.name, subject);
        } catch (RestClientException e) {
            fail("Changing config for an invalid subject should succeed");
        }
        assertEquals(FORWARD.name, confluentClient.getConfig(subject).getCompatibilityLevel(),
                "New compatibility level for this subject should be forward");
    }

    @Test
    public void testSubjectConfigChange() throws Exception {
        String subject = "testSubjectConfigChange";
        assertEquals(NONE.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "Default compatibility level should be none for this test instance");

        // change subject compatibility to forward
        confluentClient.updateCompatibility(CompatibilityLevel.FORWARD.name, subject);

        assertEquals(NONE.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "Global compatibility level should remain none for this test instance");

        assertEquals(FORWARD.name, confluentClient.getConfig(subject).getCompatibilityLevel(),
                "New compatibility level for this subject should be forward");
    }

    @Test
    public void testGlobalConfigChange() throws Exception {
        assertEquals(NONE.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "Default compatibility level should be none for this test instance");

        // change subject compatibility to forward
        confluentClient.updateCompatibility(CompatibilityLevel.FORWARD.name, null);
        assertEquals(FORWARD.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "New Global compatibility level should be forward");

        // change subject compatibility to backward
        confluentClient.updateCompatibility(BACKWARD.name, null);
        assertEquals(BACKWARD.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "New Global compatibility level should be backward");
    }

    @Test
    public void testGetSchemaNonExistingId() throws Exception {
        try {
            confluentClient.getId(Integer.MAX_VALUE);
            fail("Schema lookup by missing id should fail with " + ErrorCode.SCHEMA_NOT_FOUND.value()
                    + " (schema not found)");
        } catch (RestClientException rce) {
            // this is expected.
            assertEquals(ErrorCode.SCHEMA_NOT_FOUND.value(), rce.getErrorCode(),
                    "Should get a 404 status for non-existing id");
        }
    }

    @Test
    public void testGetSchemaTypes() throws Exception {
        assertEquals(new HashSet<>(Arrays.asList("AVRO", "JSON", "PROTOBUF")),
                new HashSet<>(confluentClient.getSchemaTypes()));
    }

    @Test
    public void testListVersionsNonExistingSubject() throws Exception {
        try {
            confluentClient.getAllVersions("Invalid");
            fail("Getting all versions of missing subject should fail with "
                    + ErrorCode.SUBJECT_NOT_FOUND.value() + " (subject not found)");
        } catch (RestClientException rce) {
            // this is expected.
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(),
                    "Should get a 404 status for non-existing subject");
        }
    }

    @Test
    public void testGetVersionNonExistentSubject() throws Exception {
        // test getVersion on a non-existing subject
        try {
            confluentClient.getVersion("non-existing-subject", 1);
            fail("Getting version of missing subject should fail with " + ErrorCode.SUBJECT_NOT_FOUND.value()
                    + " (subject not found)");
        } catch (RestClientException e) {
            // this is expected.
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), e.getErrorCode(),
                    "Unregistered subject shouldn't be found in getVersion()");
        }
    }

    @Test
    public void testGetNonExistingVersion() throws Exception {
        // test getVersion on a non-existing version
        String schema = ConfluentTestUtils.getRandomCanonicalAvroString(1).get(0);
        String subject = "test";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schema, subject);
        try {
            confluentClient.getVersion(subject, 200);
            fail("Getting unregistered version should fail with " + ErrorCode.VERSION_NOT_FOUND.value()
                    + " (version not found)");
        } catch (RestClientException e) {
            // this is expected.
            assertEquals(ErrorCode.VERSION_NOT_FOUND.value(), e.getErrorCode(),
                    "Unregistered version shouldn't be found");
        }
    }

    @Test
    public void testGetInvalidVersion() throws Exception {
        // test getVersion on a non-existing version
        String schema = ConfluentTestUtils.getRandomCanonicalAvroString(1).get(0);
        String subject = "test";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schema, subject);
        try {
            confluentClient.getVersion(subject, 0);
            fail("Getting invalid version should fail with " + ErrorCode.INVALID_VERSION
                    + " (invalid version)");
        } catch (RestClientException e) {
            // this is expected.
            assertEquals(ErrorCode.VERSION_NOT_FOUND.value(), e.getErrorCode(),
                    "Invalid version shouldn't be found");
        }
    }

    @Test
    public void testGetVersion() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject = "test";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);

        assertEquals(schemas.get(0), confluentClient.getVersion(subject, 1).getSchema(),
                "Version 1 schema should match");

        assertEquals(schemas.get(1), confluentClient.getVersion(subject, 2).getSchema(),
                "Version 2 schema should match");
        assertEquals(schemas.get(1), confluentClient.getLatestVersion(subject).getSchema(),
                "Latest schema should be the same as version 2");
    }

    @Test
    public void testGetLatestVersionSchemaOnly() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject = "test";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);

        assertEquals(schemas.get(1), confluentClient.getLatestVersionSchemaOnly(subject),
                "Latest schema should be the same as version 2");
    }

    @Test
    public void testGetVersionSchemaOnly() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(1);
        String subject = "test";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);

        assertEquals(schemas.get(0), confluentClient.getVersionSchemaOnly(subject, 1),
                "Retrieved schema should be the same as version 1");
    }

    @Test
    public void testSchemaReferences() throws Exception {
        List<String> schemas = ConfluentTestUtils.getAvroSchemaWithReferences();
        String subject = "testSchemaReferences";
        String referrer = "testSchemaReferencesReferer";

        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);

        RegisterSchemaRequest request = new RegisterSchemaRequest();
        request.setSchema(schemas.get(1));
        SchemaReference ref = new SchemaReference("otherns.Subrecord", subject, 1);
        request.setReferences(Collections.singletonList(ref));
        int registeredId = confluentClient.registerSchema(request, referrer, false).getId();

        SchemaString schemaString = confluentClient.getId(registeredId);
        // the newly registered schema should be immediately readable on the leader
        assertEquals(schemas.get(1), schemaString.getSchemaString(), "Registered schema should be found");

        assertEquals(Collections.singletonList(ref), schemaString.getReferences(),
                "Schema references should be found");

        List<Integer> refs = confluentClient.getReferencedBy(subject, 1);
        assertEquals(registeredId, refs.get(0).intValue());

        try {
            confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject,
                    String.valueOf(1));
            fail("Deleting reference should fail with " + ErrorCode.REFERENCE_EXISTS.value());
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.REFERENCE_EXISTS.value(), rce.getErrorCode(), "Reference found");
        }

        assertEquals((Integer) 1,
                confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, referrer, "1"));

        assertEquals((Integer) 1,
                confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "1"));
    }

    @Test
    public void testSchemaReferencesMultipleLevels() throws Exception {
        String root = """
                ["myavro.BudgetDecreased","myavro.BudgetUpdated"]""";

        String ref1 = """
                {
                  "type" : "record",
                  "name" : "BudgetDecreased",
                  "namespace" : "myavro",
                  "fields" : [ {
                    "name" : "buyerId",
                    "type" : "long"
                  }, {
                    "name" : "currency",
                    "type" : {
                      "type" : "myavro.currencies.Currency"    }
                  }, {
                    "name" : "amount",
                    "type" : "double"
                  } ]
                }""";

        String ref2 = """
                {
                  "type" : "record",
                  "name" : "BudgetUpdated",
                  "namespace" : "myavro",
                  "fields" : [ {
                    "name" : "buyerId",
                    "type" : "long"
                  }, {
                    "name" : "currency",
                    "type" : {
                      "type" : "myavro.currencies.Currency"    }
                  }, {
                    "name" : "updatedValue",
                    "type" : "double"
                  } ]
                }""";

        String sharedRef = """
                {
                      "type" : "enum",
                      "name" : "Currency",
                      "namespace" : "myavro.currencies",
                      "symbols" : [ "EUR", "USD" ]
                    }
                """;

        ConfluentTestUtils.registerAndVerifySchema(confluentClient,
                new AvroSchema(sharedRef).canonicalString(), "shared");

        RegisterSchemaRequest request = new RegisterSchemaRequest();
        request.setSchema(ref1);
        SchemaReference ref = new SchemaReference("myavro.currencies.Currency", "shared", 1);
        request.setReferences(Collections.singletonList(ref));
        confluentClient.registerSchema(request, "ref1", false);

        request = new RegisterSchemaRequest();
        request.setSchema(ref2);
        ref = new SchemaReference("myavro.currencies.Currency", "shared", 1);
        request.setReferences(Collections.singletonList(ref));
        confluentClient.registerSchema(request, "ref2", false);

        request = new RegisterSchemaRequest();
        request.setSchema(root);
        SchemaReference r1 = new SchemaReference("myavro.BudgetDecreased", "ref1", 1);
        SchemaReference r2 = new SchemaReference("myavro.BudgetUpdated", "ref2", 1);
        request.setReferences(Arrays.asList(r1, r2));
        int registeredSchema = confluentClient.registerSchema(request, "root", false).getId();

        SchemaString schemaString = confluentClient.getId(registeredSchema);
        // the newly registered schema should be immediately readable on the leader
        assertEquals(root, schemaString.getSchemaString(), "Registered schema should be found");

        assertEquals(Arrays.asList(r1, r2), schemaString.getReferences(),
                "Schema references should be found");
    }

    @Test
    public void testSchemaMissingReferences() {
        List<String> schemas = ConfluentTestUtils.getAvroSchemaWithReferences();

        RegisterSchemaRequest request = new RegisterSchemaRequest();
        request.setSchema(schemas.get(1));
        request.setReferences(Collections.emptyList());

        assertThrows(RestClientException.class,
                () -> confluentClient.registerSchema(request, "referrer", false));
    }

    @Test
    public void testSchemaNormalization() throws Exception {
        String subject1 = "testSchemaNormalization";

        String reference1 = "{\"type\":\"record\"," + "\"name\":\"Subrecord1\","
                + "\"namespace\":\"otherns\"," + "\"fields\":"
                + "[{\"name\":\"field1\",\"type\":\"string\"}]}";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, reference1, "ref1");
        String reference2 = "{\"type\":\"record\"," + "\"name\":\"Subrecord2\","
                + "\"namespace\":\"otherns\"," + "\"fields\":"
                + "[{\"name\":\"field2\",\"type\":\"string\"}]}";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, reference2, "ref2");

        SchemaReference ref1 = new SchemaReference("otherns.Subrecord1", "ref1", 1);
        SchemaReference ref2 = new SchemaReference("otherns.Subrecord2", "ref2", 1);

        // Two versions of same schema, the second one with extra spaces and line breaks
        String schemaString1 = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":{\"type\":\"int\"},\"name\":\"field0" + "\"},"
                + "{\"name\":\"field1\",\"type\":\"otherns.Subrecord1\"},"
                + "{\"name\":\"field2\",\"type\":\"otherns.Subrecord2\"}" + "],"
                + "\"extraMetadata\": {\"a\": 1, \"b\": 2}" + "}";
        String schemaString2 = "{\"type\":\"record\",\n" + "\"name\":\"myrecord\",\n" + "\"fields\":"
                + "[{\"type\":{\"type\":\"int\"},\"name\":\"field0" + "\"},\n"
                + "{\"name\":\"field1\",\"type\":\"otherns.Subrecord1\"},"
                + "{\"name\":\"field2\",\"type\":\"otherns.Subrecord2\"}" + "],"
                + "\"extraMetadata\": {\"a\": 1, \"b\": 2}" + "}";

        RegisterSchemaRequest registerRequest = new RegisterSchemaRequest();
        registerRequest.setSchema(schemaString1);
        registerRequest.setReferences(Arrays.asList(ref1, ref2));
        confluentClient.registerSchema(registerRequest, subject1, true);

        int versionOfRegisteredSchema1Subject1 = confluentClient
                .lookUpSubjectVersion(registerRequest, subject1, false, false).getVersion();

        // send schema with all references resolved
        RegisterSchemaRequest lookUpRequest = new RegisterSchemaRequest();
        org.apache.avro.Schema.Parser parser = new org.apache.avro.Schema.Parser();
        parser.parse(reference1);
        parser.parse(reference2);
        AvroSchema resolvedSchema = new AvroSchema(parser.parse(schemaString2));
        lookUpRequest.setSchema(resolvedSchema.canonicalString());
        versionOfRegisteredSchema1Subject1 = confluentClient
                .lookUpSubjectVersion(lookUpRequest, subject1, true, false).getVersion();
        assertEquals(1, versionOfRegisteredSchema1Subject1,
                "1st schema under subject1 should have version 1");

        String recordInvalidDefaultSchema = "{\"namespace\": \"namespace\",\n" + " \"type\": \"record\",\n"
                + " \"name\": \"test\",\n" + " \"fields\": [\n"
                + "     {\"name\": \"string_default\", \"type\": \"string\", \"default\": null}\n" + "]\n"
                + "}";
        registerRequest = new RegisterSchemaRequest();
        registerRequest.setSchema(recordInvalidDefaultSchema);
        try {
            confluentClient.registerSchema(registerRequest, subject1, true);
            fail("Registering bad schema should fail with " + ErrorCode.INVALID_SCHEMA.value());
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.INVALID_SCHEMA.value(), rce.getErrorCode(), "Invalid schema");
        }
    }

    @Test
    public void testBad() throws Exception {
        String subject1 = "testBad";
        List<String> allSubjects = new ArrayList<>();

        // test getAllSubjects with no existing data
        assertEquals(allSubjects, confluentClient.getAllSubjects(),
                "Getting all subjects should return empty");

        try {
            ConfluentTestUtils.registerAndVerifySchema(confluentClient, ConfluentTestUtils.getBadSchema(),
                    subject1);
            fail("Registering bad schema should fail with " + ErrorCode.INVALID_SCHEMA.value());
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.INVALID_SCHEMA.value(), rce.getErrorCode(), "Invalid schema");
        }

        try {
            ConfluentTestUtils.registerAndVerifySchema(confluentClient,
                    ConfluentTestUtils.getRandomCanonicalAvroString(1).get(0),
                    List.of(new SchemaReference("bad", "bad", 100)), subject1);
            fail("Registering bad reference should fail with " + ErrorCode.INVALID_SCHEMA.value());
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.INVALID_SCHEMA.value(), rce.getErrorCode(), "Invalid schema");
        }

        // test getAllSubjects with existing data
        assertEquals(allSubjects, confluentClient.getAllSubjects(),
                "Getting all subjects should match all registered subjects");
    }

    @Test
    public void testLookUpSchemaUnderNonExistentSubject() throws Exception {
        String schema = ConfluentTestUtils.getRandomCanonicalAvroString(1).get(0);
        try {
            confluentClient.lookUpSubjectVersion(schema, "non-existent-subject");
            fail("Looking up schema under missing subject should fail with "
                    + ErrorCode.SUBJECT_NOT_FOUND.value() + " (subject not found)");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(), "Subject not found");
        }
    }

    @Test
    public void testLookUpNonExistentSchemaUnderSubject() throws Exception {
        String subject = "test";
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        confluentClient.updateCompatibility(CompatibilityLevel.NONE.name, subject);

        try {
            confluentClient.lookUpSubjectVersion(schemas.get(1), subject);
            fail("Looking up missing schema under subject should fail with "
                    + ErrorCode.SCHEMA_NOT_FOUND.value() + " (schema not found)");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SCHEMA_NOT_FOUND.value(), rce.getErrorCode(), "Schema not found");
        }
    }

    @Test
    public void testGetVersionsAssociatedWithSchemaId() throws Exception {
        String subject1 = "testGetVersionsAssociatedWithSchemaId1";
        String subject2 = "testGetVersionsAssociatedWithSchemaId2";

        String schema = ConfluentTestUtils.getRandomCanonicalAvroString(1).get(0);

        int registeredSchemaId = confluentClient.registerSchema(schema, subject1);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schema, subject1);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schema, subject2);

        List<SubjectVersion> associatedSubjects = confluentClient.getAllVersionsById(registeredSchemaId);
        assertEquals(associatedSubjects.size(), 2);
        assertTrue(associatedSubjects.contains(new SubjectVersion(subject1, 1)));
        assertTrue(associatedSubjects.contains(new SubjectVersion(subject2, 1)));

        assertEquals((Integer) 1,
                confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject2, "1"),
                "Deleting Schema Version Success");

        associatedSubjects = confluentClient.getAllVersionsById(registeredSchemaId);
        assertEquals(associatedSubjects.size(), 1);
        assertTrue(associatedSubjects.contains(new SubjectVersion(subject1, 1)));

        associatedSubjects = confluentClient.getAllVersionsById(RestService.DEFAULT_REQUEST_PROPERTIES,
                registeredSchemaId, null, true);
        assertEquals(associatedSubjects.size(), 2);
        assertTrue(associatedSubjects.contains(new SubjectVersion(subject1, 1)));
        assertTrue(associatedSubjects.contains(new SubjectVersion(subject2, 1)));
    }

    @Test
    public void testCompatibilityNonExistentVersion() throws Exception {
        String schema = ConfluentTestUtils.getRandomCanonicalAvroString(1).get(0);
        String subject = "testCompatibilityNonExistentVersion";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schema, subject);
        try {
            confluentClient.testCompatibility(schema, subject, "100");
            fail("Testing compatibility for missing version should fail with "
                    + ErrorCode.VERSION_NOT_FOUND.value() + " (version not found)");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.VERSION_NOT_FOUND.value(), rce.getErrorCode(), "Version not found");
        }
    }

    @Test
    public void testCompatibilityInvalidVersion() throws Exception {
        String schema = ConfluentTestUtils.getRandomCanonicalAvroString(1).get(0);
        String subject = "test";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schema, subject);
        try {
            confluentClient.testCompatibility(schema, subject, "earliest");
            fail("Testing compatibility for invalid version should fail with "
                    + ErrorCode.VERSION_NOT_FOUND.value() + " (version not found)");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.VERSION_NOT_FOUND.value(), rce.getErrorCode(), "Version not found");
        }
    }

    @Test
    public void testGetConfigNonExistentSubject() throws Exception {
        try {
            confluentClient.getConfig("non-existent-subject");
            fail("Getting the configuration of a missing subject should fail with "
                    + ErrorCode.SUBJECT_NOT_FOUND.value() + " error code (subject not found)");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(), "Subject not found");
        }
    }

    @Test
    public void testCanonicalization() throws Exception {
        // schema string with extra white space
        String schema = "{   \"type\":   \"string\"}";
        String subject = "test";

        int id = confluentClient.registerSchema(schema, subject);

        assertEquals(id, confluentClient.registerSchema(schema, subject),
                "Registering the same schema should get back the same id");

        assertEquals(id, confluentClient.lookUpSubjectVersion(schema, subject).getId().intValue(),
                "Lookup the same schema should get back the same id");
    }

    @Test
    public void testDeleteSchemaVersionBasic() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject = "testDeleteSchemaVersionBasic";

        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);

        assertEquals((Integer) 2,
                confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "2"),
                "Deleting Schema Version Success");

        assertEquals(Collections.singletonList(1), confluentClient.getAllVersions(subject));

        try {
            confluentClient.getVersion(subject, 2);
            fail(String.format("Getting Version %s for subject %s should fail with %s", "2", subject,
                    ErrorCode.VERSION_NOT_FOUND.value()));
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.VERSION_NOT_FOUND.value(), rce.getErrorCode(), "Version not found");
        }
        try {
            RegisterSchemaRequest request = new RegisterSchemaRequest();
            request.setSchema(schemas.get(1));
            confluentClient.lookUpSubjectVersion(RestService.DEFAULT_REQUEST_PROPERTIES, request, subject,
                    false, false);
            fail(String.format("Lookup Subject Version %s for subject %s should fail with %s", "2", subject,
                    ErrorCode.SCHEMA_NOT_FOUND.value()));
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SCHEMA_NOT_FOUND.value(), rce.getErrorCode(), "Schema not found");
        }

        assertEquals((Integer) 1, confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES,
                subject, "latest"), "Deleting Schema Version Success");
        try {
            List<Integer> versions = confluentClient.getAllVersions(subject);
            fail("Getting all versions from non-existing subject1 should fail with "
                    + ErrorCode.SUBJECT_NOT_FOUND.value() + " (subject not found). Got " + versions);
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(),
                    "Should get a 404 status for non-existing subject");
        }

        // re-register twice and versions should be same
        for (int i = 0; i < 2; i++) {
            ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
            assertEquals(Collections.singletonList(3), confluentClient.getAllVersions(subject));
        }

    }

    @Test
    public void testDeleteSchemaVersionPermanent() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject = "test";

        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);

        // permanent delete without soft delete first
        try {
            confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "2", true);
            fail("Permanent deleting first time should throw schemaVersionNotSoftDeletedException");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SCHEMA_VERSION_NOT_SOFT_DELETED.value(), rce.getErrorCode(),
                    "Schema version must be soft deleted first");
        }

        // soft delete
        assertEquals((Integer) 2,
                confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "2"),
                "Deleting Schema Version Success");

        assertEquals(Collections.singletonList(1), confluentClient.getAllVersions(subject));
        assertEquals(Arrays.asList(1, 2),
                confluentClient.getAllVersions(RestService.DEFAULT_REQUEST_PROPERTIES, subject, true));
        // soft delete again
        try {
            confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "2");
            fail("Soft deleting second time should throw schemaVersionSoftDeletedException");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SCHEMA_VERSION_SOFT_DELETED.value(), rce.getErrorCode(),
                    "Schema version already soft deleted");
        }

        try {
            confluentClient.getVersion(subject, 2);
            fail(String.format("Getting Version %s for subject %s should fail with %s", "2", subject,
                    ErrorCode.VERSION_NOT_FOUND.value()));
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.VERSION_NOT_FOUND.value(), rce.getErrorCode(), "Version not found");
        }

        io.confluent.kafka.schemaregistry.client.rest.entities.Schema schema = confluentClient
                .getVersion(subject, 2, true);
        assertEquals((Integer) 2, schema.getVersion(), "Lookup Version Match");

        try {
            RegisterSchemaRequest request = new RegisterSchemaRequest();
            request.setSchema(schemas.get(1));
            confluentClient.lookUpSubjectVersion(RestService.DEFAULT_REQUEST_PROPERTIES, request, subject,
                    false, false);
            fail(String.format("Lookup Subject Version %s for subject %s should fail with %s", "2", subject,
                    ErrorCode.SCHEMA_NOT_FOUND.value()));
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SCHEMA_NOT_FOUND.value(), rce.getErrorCode(), "Schema not found");
        }
        // permanent delete
        assertEquals((Integer) 2, confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES,
                subject, "2", true), "Deleting Schema Version Success");
        // GET after permanent delete should give exception
        try {
            confluentClient.getVersion(subject, 2, true);
            fail(String.format("Getting Version %s for subject %s should fail with %s", "2", subject,
                    ErrorCode.VERSION_NOT_FOUND.value()));
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.VERSION_NOT_FOUND.value(), rce.getErrorCode(), "Version not found");
        }
        // permanent delete again
        try {
            confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "2", true);
            fail(String.format("Getting Version %s for subject %s should fail with %s", "2", subject,
                    ErrorCode.VERSION_NOT_FOUND.value()));
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.VERSION_NOT_FOUND.value(), rce.getErrorCode(), "Version not found");
        }

        assertEquals((Integer) 1, confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES,
                subject, "latest"), "Deleting Schema Version Success");

        try {
            List<Integer> versions = confluentClient.getAllVersions(subject);
            fail("Getting all versions from non-existing subject1 should fail with "
                    + ErrorCode.SUBJECT_NOT_FOUND.value() + " (subject not found). Got " + versions);
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(),
                    "Should get a 404 status for non-existing subject");
        }

        // re-register twice and versions should be same
        // after permanent delete of 2, the new version coming up will be 2
        for (int i = 0; i < 2; i++) {
            ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
            assertEquals(Collections.singletonList(2), confluentClient.getAllVersions(subject));
        }

    }

    @Test
    public void testDeleteSchemaVersionInvalidSubject() throws Exception {
        try {
            String subject = "testDeleteSchemaVersionInvalidSubject";
            confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "1");
            fail("Deleting a non existent subject version should fail with "
                    + ErrorCode.SUBJECT_NOT_FOUND.value() + " error code (subject not found)");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(), "Subject not found");
        }
    }

    @Test
    public void testDeleteLatestVersion() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(3);
        String subject = "testDeleteLatestVersion";

        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);

        assertEquals((Integer) 2, confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES,
                subject, "latest"), "Deleting Schema Version Success");

        io.confluent.kafka.schemaregistry.client.rest.entities.Schema schema = confluentClient
                .getLatestVersion(subject);
        assertEquals(schemas.get(0), schema.getSchema(), "Latest Version Schema");

        assertEquals((Integer) 1, confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES,
                subject, "latest"), "Deleting Schema Version Success");
        try {
            confluentClient.getLatestVersion(subject);
            fail("Getting latest versions from non-existing subject should fail with "
                    + ErrorCode.SUBJECT_NOT_FOUND.value() + " (subject not found).");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(),
                    "Should get a 404 status for non-existing subject");
        }

        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(2), subject);
        assertEquals(schemas.get(2), confluentClient.getLatestVersion(subject).getSchema(),
                "Latest version available after subject re-registration");
    }

    @Test
    public void testGetLatestVersionNonExistentSubject() throws Exception {
        String subject = "non_existent_subject";

        try {
            confluentClient.getLatestVersion(subject);
            fail("Getting latest versions from non-existing subject should fail with "
                    + ErrorCode.SUBJECT_NOT_FOUND.value() + " (subject not found).");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(),
                    "Should get a 404 status for non-existing subject");
        }
    }

    @Test
    public void testGetLatestVersionDeleteOlder() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject = "test";

        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);

        assertEquals(schemas.get(1), confluentClient.getLatestVersion(subject).getSchema(),
                "Latest Version Schema");

        assertEquals((Integer) 1,
                confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "1"),
                "Deleting Schema Older Version Success");
        assertEquals(schemas.get(1), confluentClient.getLatestVersion(subject).getSchema(),
                "Latest Version Schema Still Same");
    }

    @Test
    public void testDeleteInvalidVersion() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(1);
        String subject = "test";

        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        try {
            confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "2");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.VERSION_NOT_FOUND.value(), rce.getErrorCode(),
                    "Should get a 404 status for non-existing subject version");
        }

    }

    @Test
    public void testDeleteWithLookup() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject = "testDeleteWithLookup";

        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);
        assertEquals((Integer) 1,
                confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "1"),
                "Deleting Schema Version Success");
        try {
            confluentClient.lookUpSubjectVersion(schemas.get(0), subject, false);
            fail(String.format("Lookup Subject Version %s for subject %s should fail with %s", "2", subject,
                    ErrorCode.SCHEMA_NOT_FOUND.value()));
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SCHEMA_NOT_FOUND.value(), rce.getErrorCode(), "Schema not found");
        }
        // verify deleted schema
        io.confluent.kafka.schemaregistry.client.rest.entities.Schema schema = confluentClient
                .lookUpSubjectVersion(schemas.get(0), subject, true);
        assertEquals((Integer) 1, schema.getVersion(), "Lookup Version Match");

        // re-register schema again and verify we get latest version
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        schema = confluentClient.lookUpSubjectVersion(schemas.get(0), subject, true);
        assertEquals((Integer) 3, schema.getVersion(), "Lookup Version Match");
        schema = confluentClient.lookUpSubjectVersion(schemas.get(1), subject, false);
        assertEquals((Integer) 2, schema.getVersion(), "Lookup Version Match");
    }

    @Test
    public void testIncompatibleSchemaLookupBySubjectAfterDelete() throws Exception {
        String subject = "testIncompatibleSchemaLookupBySubjectAfterDelete";

        // Make two incompatible schemas - field 'g' has different types
        String schema1String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":" + "\"f" + "\"}]}";
        String schema1 = new AvroSchema(schema1String).canonicalString();

        String wrongSchema2String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":" + "\"f" + "\"}," + "{\"type\":\"string\",\"name\":"
                + "\"g\" , \"default\":\"d\"}" + "]}";
        String wrongSchema2 = new AvroSchema(wrongSchema2String).canonicalString();

        String correctSchema2String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":" + "\"f" + "\"}," + "{\"type\":\"int\",\"name\":"
                + "\"g\" , \"default\":0}" + "]}";
        String correctSchema2 = new AvroSchema(correctSchema2String).canonicalString();
        // ensure registering incompatible schemas will raise an error
        confluentClient.updateCompatibility(CompatibilityLevel.BACKWARD.name, subject);

        // test that compatibility check for incompatible schema returns false and the appropriate
        // error response from Avro
        confluentClient.registerSchema(schema1, subject, true);

        boolean isCompatible = confluentClient.testCompatibility(wrongSchema2, subject, "latest").isEmpty();
        assertTrue(isCompatible, "Schema should be compatible with specified version");

        confluentClient.registerSchema(wrongSchema2, subject, true);

        isCompatible = confluentClient.testCompatibility(correctSchema2, subject, "latest").isEmpty();
        assertFalse(isCompatible, "Schema should be incompatible with specified version");
        try {
            confluentClient.registerSchema(correctSchema2, subject);
            fail("Schema should be Incompatible");
        } catch (RestClientException rce) {
            assertEquals(HTTP_CONFLICT, rce.getErrorCode(), "Incompatible Schema");
        }

        confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "latest");
        isCompatible = confluentClient.testCompatibility(correctSchema2, subject, "latest").isEmpty();
        assertTrue(isCompatible, "Schema should be compatible with specified version");

        confluentClient.registerSchema(correctSchema2, subject, true);

        assertEquals((Integer) 3,
                confluentClient.lookUpSubjectVersion(correctSchema2String, subject, true, false).getVersion(),
                "Version is same");

    }

    @Test
    public void testSubjectCompatibilityAfterDeletingAllVersions() throws Exception {
        String subject = "testSubjectCompatibilityAfterDeletingAllVersions";

        String schema1String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":" + "\"f" + "\"}]}";
        String schema1 = new AvroSchema(schema1String).canonicalString();

        String schema2String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":" + "\"f" + "\"}," + "{\"type\":\"string\",\"name\":"
                + "\"g\" , \"default\":\"d\"}" + "]}";
        String schema2 = new AvroSchema(schema2String).canonicalString();

        confluentClient.updateCompatibility(CompatibilityLevel.FULL.name, null);
        confluentClient.updateCompatibility(CompatibilityLevel.BACKWARD.name, subject);

        confluentClient.registerSchema(schema1, subject, true);
        confluentClient.registerSchema(schema2, subject);

        confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "1");
        assertEquals(BACKWARD.name, confluentClient.getConfig(subject).getCompatibilityLevel(),
                "Compatibility Level Exists");
        assertEquals(FULL.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "Top Compatibility Level Exists");
        confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject, "2");
        try {
            confluentClient.getConfig(subject);
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_COMPATIBILITY_NOT_CONFIGURED.value(), rce.getErrorCode(),
                    "Compatibility Level doesn't exist");
        }
        assertEquals(FULL.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "Top Compatibility Level Exists");

    }

    @Test
    public void testListSubjects() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject1 = "test1";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject1);
        String subject2 = "test2";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject2);

        List<String> expectedResponse = new ArrayList<>();
        expectedResponse.add(subject1);
        expectedResponse.add(subject2);
        assertEquals(expectedResponse, confluentClient.getAllSubjects(), "Current Subjects");
        List<Integer> deletedResponse = new ArrayList<>();
        deletedResponse.add(1);
        assertEquals(deletedResponse,
                confluentClient.deleteSubject(RestService.DEFAULT_REQUEST_PROPERTIES, subject2),
                "Versions Deleted Match");

        expectedResponse = new ArrayList<>();
        expectedResponse.add(subject1);
        assertEquals(expectedResponse, confluentClient.getAllSubjects(), "Current Subjects");

        expectedResponse = new ArrayList<>();
        expectedResponse.add(subject1);
        expectedResponse.add(subject2);
        assertEquals(expectedResponse, confluentClient.getAllSubjects(true), "Current Subjects");

        assertEquals(deletedResponse,
                confluentClient.deleteSubject(RestService.DEFAULT_REQUEST_PROPERTIES, subject2, true),
                "Versions Deleted Match");

        expectedResponse = new ArrayList<>();
        expectedResponse.add(subject1);
        assertEquals(expectedResponse, confluentClient.getAllSubjects(), "Current Subjects");
    }

    @Test
    public void testListSoftDeletedSubjectsAndSchemas() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(3);
        String subject1 = "test1";
        String subject2 = "test2";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject1);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject1);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(2), subject2);

        assertEquals((Integer) 1,
                confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject1, "1"));
        assertEquals((Integer) 1,
                confluentClient.deleteSchemaVersion(RestService.DEFAULT_REQUEST_PROPERTIES, subject2, "1"));

        assertEquals(Collections.singletonList(2), confluentClient.getAllVersions(subject1),
                "List All Versions Match");
        assertEquals(Arrays.asList(1, 2),
                confluentClient.getAllVersions(RestService.DEFAULT_REQUEST_PROPERTIES, subject1, true),
                "List All Versions Include deleted Match");

        assertEquals(Collections.singletonList(subject1), confluentClient.getAllSubjects(),
                "List All Subjects Match");
        assertEquals(Arrays.asList(subject1, subject2), confluentClient.getAllSubjects(true),
                "List All Subjects Include deleted Match");
    }

    @Test
    public void testDeleteSubjectBasic() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject = "testDeleteSubjectBasic";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);
        List<Integer> expectedResponse = new ArrayList<>();
        expectedResponse.add(1);
        expectedResponse.add(2);
        assertEquals(expectedResponse,
                confluentClient.deleteSubject(RestService.DEFAULT_REQUEST_PROPERTIES, subject),
                "Versions Deleted Match");
        try {
            confluentClient.getLatestVersion(subject);
            fail(String.format("Subject %s should not be found", subject));
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(), "Subject Not Found");
        }

    }

    @Test
    public void testDeleteSubjectException() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject = "testDeleteSubjectException";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);
        List<Integer> expectedResponse = new ArrayList<>();
        expectedResponse.add(1);
        expectedResponse.add(2);
        assertEquals(expectedResponse,
                confluentClient.deleteSubject(RestService.DEFAULT_REQUEST_PROPERTIES, subject),
                "Versions Deleted Match");

        io.confluent.kafka.schemaregistry.client.rest.entities.Schema schema = confluentClient
                .lookUpSubjectVersion(schemas.get(0), subject, true);
        assertEquals(1, (long) schema.getVersion());
        schema = confluentClient.lookUpSubjectVersion(schemas.get(1), subject, true);
        assertEquals(2, (long) schema.getVersion());

        try {
            confluentClient.deleteSubject(RestService.DEFAULT_REQUEST_PROPERTIES, subject);
            fail(String.format("Subject %s should not be found", subject));
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_SOFT_DELETED.value(), rce.getErrorCode(),
                    "Subject exists in soft deleted format.");
        }
    }

    @Test
    public void testDeleteSubjectPermanent() throws Exception {
        List<String> schemas = ConfluentTestUtils.getRandomCanonicalAvroString(2);
        String subject = "testDeleteSubjectPermanent";
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(0), subject);
        ConfluentTestUtils.registerAndVerifySchema(confluentClient, schemas.get(1), subject);
        List<Integer> expectedResponse = new ArrayList<>();
        expectedResponse.add(1);
        expectedResponse.add(2);

        try {
            confluentClient.deleteSubject(RestService.DEFAULT_REQUEST_PROPERTIES, subject, true);
            fail("Delete permanent should not succeed");
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_NOT_SOFT_DELETED.value(), rce.getErrorCode(),
                    "Subject '%s' was not deleted first before permanent delete");
        }

        assertEquals(expectedResponse,
                confluentClient.deleteSubject(RestService.DEFAULT_REQUEST_PROPERTIES, subject),
                "Versions Deleted Match");

        io.confluent.kafka.schemaregistry.client.rest.entities.Schema schema = confluentClient
                .lookUpSubjectVersion(schemas.get(0), subject, true);
        assertEquals(1, (long) schema.getVersion());
        schema = confluentClient.lookUpSubjectVersion(schemas.get(1), subject, true);
        assertEquals(2, (long) schema.getVersion());

        assertEquals(expectedResponse,
                confluentClient.deleteSubject(RestService.DEFAULT_REQUEST_PROPERTIES, subject, true),
                "Versions Deleted Match");
        for (Integer i : expectedResponse) {
            try {
                confluentClient.lookUpSubjectVersion(schemas.get(0), subject, false);
                fail(String.format("Subject %s should not be found", subject));
            } catch (RestClientException rce) {
                assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(), "Subject Not Found");
            }

            try {
                confluentClient.lookUpSubjectVersion(schemas.get(i - 1), subject, true);
                fail(String.format("Subject %s should not be found", subject));
            } catch (RestClientException rce) {
                assertEquals(ErrorCode.SUBJECT_NOT_FOUND.value(), rce.getErrorCode(), "Subject Not Found");
            }
        }
    }

    @Test
    public void testSubjectCompatibilityAfterDeletingSubject() throws Exception {
        String subject = "testSubjectCompatibilityAfterDeletingSubject";

        String schema1String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":" + "\"f" + "\"}]}";
        String schema1 = new AvroSchema(schema1String).canonicalString();

        String schema2String = "{\"type\":\"record\"," + "\"name\":\"myrecord\"," + "\"fields\":"
                + "[{\"type\":\"string\",\"name\":" + "\"f" + "\"}," + "{\"type\":\"string\",\"name\":"
                + "\"g\" , \"default\":\"d\"}" + "]}";
        String schema2 = new AvroSchema(schema2String).canonicalString();

        confluentClient.updateCompatibility(CompatibilityLevel.FULL.name, null);
        confluentClient.updateCompatibility(CompatibilityLevel.BACKWARD.name, subject);

        confluentClient.registerSchema(schema1, subject, true);
        confluentClient.registerSchema(schema2, subject, true);

        confluentClient.deleteSubject(RestService.DEFAULT_REQUEST_PROPERTIES, subject);
        try {
            confluentClient.getConfig(subject);
        } catch (RestClientException rce) {
            assertEquals(ErrorCode.SUBJECT_COMPATIBILITY_NOT_CONFIGURED.value(), rce.getErrorCode(),
                    "Compatibility Level doesn't exist");
        }
        assertEquals(FULL.name, confluentClient.getConfig(null).getCompatibilityLevel(),
                "Top Compatibility Level Exists");

    }
}
