@Override
    public Schema readSchema(InputStream in) throws IOException {
        DataInputStream dataInputStream = new DataInputStream(in);

        if (dataInputStream.readByte() != 0) {
            throw new IOException("Unknown data format. Magic number does not match");
        } else {
            int schemaId = dataInputStream.readInt();

            try {
                return schemaRegistryClient.getById(schemaId);
            } catch (RestClientException e) {
                throw new IOException(
                        format("Could not find schema with id %s in registry", schemaId), e);
            }
        }
    }
@Test
    void testSpecificRecordWithConfluentSchemaRegistry() throws Exception {
        MockSchemaRegistryClient client = new MockSchemaRegistryClient();

        Schema schema =
                SchemaBuilder.record("testRecord").fields().optionalString("testField").endRecord();
        int schemaId = client.register("testTopic", schema);

        ConfluentSchemaRegistryCoder registryCoder = new ConfluentSchemaRegistryCoder(client);
        ByteArrayOutputStream byteOutStream = new ByteArrayOutputStream();
        DataOutputStream dataOutputStream = new DataOutputStream(byteOutStream);
        dataOutputStream.writeByte(0);
        dataOutputStream.writeInt(schemaId);
        dataOutputStream.flush();

        ByteArrayInputStream byteInStream = new ByteArrayInputStream(byteOutStream.toByteArray());
        Schema readSchema = registryCoder.readSchema(byteInStream);

        assertThat(readSchema).isEqualTo(schema);
        assertThat(byteInStream).isEmpty();
    }