@Test public void testGetSchemaWithValidEncoding() throws IOException, SchemaNotFoundException { final SchemaAccessStrategy schemaAccessStrategy = new HortonworksEncodedSchemaReferenceStrategy(schemaRegistry); final int protocol = 1; final long schemaId = 123456; final int version = 2; try (final ByteArrayOutputStream bytesOut = new ByteArrayOutputStream(); final DataOutputStream out = new DataOutputStream(bytesOut)) { out.write(protocol); out.writeLong(schemaId); out.writeInt(version); out.flush(); try (final ByteArrayInputStream in = new ByteArrayInputStream(bytesOut.toByteArray())) { final SchemaIdentifier expectedSchemaIdentifier = SchemaIdentifier.builder() .id(schemaId) .version(version) .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[13]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 13 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int protocolVersion = bb.get(); if (protocolVersion != 1) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. The latest known Protocol is Version " + LATEST_PROTOCOL_VERSION + " but the data was encoded with version " + protocolVersion + " or was not encoded with this data format"); } final long schemaId = bb.getLong(); final int schemaVersion = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder().id(schemaId).version(schemaVersion).build(); return schemaRegistry.retrieveSchema(schemaIdentifier); }
HortonworksEncodedSchemaReferenceStrategy 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[13]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 13 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int protocolVersion = bb.get(); if (protocolVersion != 1) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. The latest known Protocol is Version " + LATEST_PROTOCOL_VERSION + " but the data was encoded with version " + protocolVersion + " or was not encoded with this data format"); } final long schemaId = bb.getLong(); final int schemaVersion = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder().id(schemaId).version(schemaVersion).build(); return schemaRegistry.retrieveSchema(schemaIdentifier); } }
HortonworksEncodedSchemaReferenceStrategy 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[13]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 13 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int protocolVersion = bb.get(); if (protocolVersion != 1) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. The latest known Protocol is Version " + LATEST_PROTOCOL_VERSION + " but the data was encoded with version " + protocolVersion + " or was not encoded with this data format"); } final long schemaId = bb.getLong(); final int schemaVersion = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder().id(schemaId).version(schemaVersion).build(); return schemaRegistry.retrieveSchema(schemaIdentifier); } HortonworksEncodedSchemaReferenceStrategy(final SchemaRegistry schemaRegistry); }
HortonworksEncodedSchemaReferenceStrategy 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[13]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 13 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int protocolVersion = bb.get(); if (protocolVersion != 1) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. The latest known Protocol is Version " + LATEST_PROTOCOL_VERSION + " but the data was encoded with version " + protocolVersion + " or was not encoded with this data format"); } final long schemaId = bb.getLong(); final int schemaVersion = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder().id(schemaId).version(schemaVersion).build(); return schemaRegistry.retrieveSchema(schemaIdentifier); } HortonworksEncodedSchemaReferenceStrategy(final SchemaRegistry schemaRegistry); @Override RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields(); }
HortonworksEncodedSchemaReferenceStrategy 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[13]; try { StreamUtils.fillBuffer(contentStream, buffer); } catch (final IOException ioe) { throw new SchemaNotFoundException("Could not read first 13 bytes from stream", ioe); } final ByteBuffer bb = ByteBuffer.wrap(buffer); final int protocolVersion = bb.get(); if (protocolVersion != 1) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. The latest known Protocol is Version " + LATEST_PROTOCOL_VERSION + " but the data was encoded with version " + protocolVersion + " or was not encoded with this data format"); } final long schemaId = bb.getLong(); final int schemaVersion = bb.getInt(); final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder().id(schemaId).version(schemaVersion).build(); return schemaRegistry.retrieveSchema(schemaIdentifier); } HortonworksEncodedSchemaReferenceStrategy(final SchemaRegistry schemaRegistry); @Override RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields();  }