@Test public void testGetSchemaWithValidEncoding() throws IOException, SchemaNotFoundException { final SchemaAccessStrategy schemaAccessStrategy = new ConfluentSchemaRegistryStrategy(schemaRegistry); final int schemaId = 123456; try (final ByteArrayOutputStream bytesOut = new ByteArrayOutputStream(); final DataOutputStream out = new DataOutputStream(bytesOut)) { out.write(0); out.writeInt(schemaId); out.flush(); try (final ByteArrayInputStream in = new ByteArrayInputStream(bytesOut.toByteArray())) { final SchemaIdentifier expectedSchemaIdentifier = SchemaIdentifier.builder() .id((long)schemaId) .version(1) .build(); when(schemaRegistry.retrieveSchema(argThat(new SchemaIdentifierMatcher(expectedSchemaIdentifier)))) .thenReturn(recordSchema); final RecordSchema retrievedSchema = schemaAccessStrategy.getSchema(Collections.emptyMap(), in, recordSchema); assertNotNull(retrievedSchema); } } }
@Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException, IOException { final byte[] buffer = new byte[5]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 5 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int magicByte = bb.get(); if (magicByte != 0) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. " + "Expected stream to begin with a Magic Byte of 0 but first byte was " + magicByte); } final int schemaId = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder() .id(Long.valueOf(schemaId)) .version(1) .build(); return schemaRegistry.retrieveSchema(schemaIdentifier); }
ConfluentSchemaRegistryStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException, IOException { final byte[] buffer = new byte[5]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 5 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int magicByte = bb.get(); if (magicByte != 0) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. " + "Expected stream to begin with a Magic Byte of 0 but first byte was " + magicByte); } final int schemaId = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder() .id(Long.valueOf(schemaId)) .version(1) .build(); return schemaRegistry.retrieveSchema(schemaIdentifier); } }
ConfluentSchemaRegistryStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException, IOException { final byte[] buffer = new byte[5]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 5 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int magicByte = bb.get(); if (magicByte != 0) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. " + "Expected stream to begin with a Magic Byte of 0 but first byte was " + magicByte); } final int schemaId = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder() .id(Long.valueOf(schemaId)) .version(1) .build(); return schemaRegistry.retrieveSchema(schemaIdentifier); } ConfluentSchemaRegistryStrategy(final SchemaRegistry schemaRegistry); }
ConfluentSchemaRegistryStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException, IOException { final byte[] buffer = new byte[5]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 5 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int magicByte = bb.get(); if (magicByte != 0) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. " + "Expected stream to begin with a Magic Byte of 0 but first byte was " + magicByte); } final int schemaId = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder() .id(Long.valueOf(schemaId)) .version(1) .build(); return schemaRegistry.retrieveSchema(schemaIdentifier); } ConfluentSchemaRegistryStrategy(final SchemaRegistry schemaRegistry); @Override RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields(); }
ConfluentSchemaRegistryStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException, IOException { final byte[] buffer = new byte[5]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 5 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int magicByte = bb.get(); if (magicByte != 0) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. " + "Expected stream to begin with a Magic Byte of 0 but first byte was " + magicByte); } final int schemaId = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder() .id(Long.valueOf(schemaId)) .version(1) .build(); return schemaRegistry.retrieveSchema(schemaIdentifier); } ConfluentSchemaRegistryStrategy(final SchemaRegistry schemaRegistry); @Override RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields();  }