### Question:
ConfluentSchemaRegistryCoder implements SchemaCoder { @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); } } } ConfluentSchemaRegistryCoder(SchemaRegistryClient schemaRegistryClient); @Override Schema readSchema(InputStream in);  }### Answer:
@Test public 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); assertEquals(schema, readSchema); assertEquals(0, byteInStream.available()); }
@Test(expected = IOException.class) public void testMagicByteVerification() throws Exception { MockSchemaRegistryClient client = new MockSchemaRegistryClient(); int schemaId = client.register("testTopic", Schema.create(Schema.Type.BOOLEAN)); ConfluentSchemaRegistryCoder coder = new ConfluentSchemaRegistryCoder(client); ByteArrayOutputStream byteOutStream = new ByteArrayOutputStream(); DataOutputStream dataOutputStream = new DataOutputStream(byteOutStream); dataOutputStream.writeByte(5); dataOutputStream.writeInt(schemaId); dataOutputStream.flush(); ByteArrayInputStream byteInStream = new ByteArrayInputStream(byteOutStream.toByteArray()); coder.readSchema(byteInStream); }