ByteBufferConverter implements Converter, HeaderConverter { @Override public byte[] fromConnectData(String topic, Schema schema, Object value) { if (schema != null && schema.type() != Schema.Type.BYTES) { assertDataException("schema", schema.type()); LOGGER.debug("Value is not of Schema.Type.BYTES, delegating to " + delegateConverter.getClass().getName()); return delegateConverter.fromConnectData(topic, schema, value); } else if (value != null && !(value instanceof ByteBuffer)) { assertDataException("value", value.getClass().getName()); LOGGER.debug("Value is not of type ByteBuffer, delegating to " + delegateConverter.getClass().getName()); return delegateConverter.fromConnectData(topic, schema, value); } return value == null ? null : ((ByteBuffer) value).array(); }  @Override ConfigDef config(); @Override void configure(Map<String, ?> configs); @Override void configure(Map<String, ?> configs, boolean isKey); @Override byte[] fromConnectData(String topic, Schema schema, Object value); @Override SchemaAndValue toConnectData(String topic, byte[] value); @Override byte[] fromConnectHeader(String topic, String headerKey, Schema schema, Object value); @Override SchemaAndValue toConnectHeader(String topic, String headerKey, byte[] value); @Override void close(); static final String DELEGATE_CONVERTER_TYPE; }
@Test public void shouldConvertFromConnectData() { byte[] bytes = converter.fromConnectData(TOPIC, Schema.BYTES_SCHEMA, ByteBuffer.wrap(SAMPLE_BYTES)); assertThat(bytes).isEqualTo(SAMPLE_BYTES); }
@Test public void shouldConvertFromConnectDataForOptionalBytesSchema() { byte[] bytes = converter.fromConnectData(TOPIC, Schema.OPTIONAL_BYTES_SCHEMA, ByteBuffer.wrap(SAMPLE_BYTES)); assertThat(bytes).isEqualTo(SAMPLE_BYTES); }
@Test public void shouldConvertFromConnectDataWithoutSchema() { byte[] bytes = converter.fromConnectData(TOPIC, null, ByteBuffer.wrap(SAMPLE_BYTES)); assertThat(bytes).isEqualTo(SAMPLE_BYTES); }
@Test public void shouldConvertNullFromConnectData() { byte[] bytes = converter.fromConnectData(TOPIC, null, null); assertThat(bytes).isNull(); }
@Test public void shouldThrowWhenConvertNonByteSchemaFromConnectData() { try { converter.fromConnectData(TOPIC, Schema.BOOLEAN_SCHEMA, null); fail("now expected exception thrown"); } catch (Exception e) { assertThat(e).isExactlyInstanceOf(DataException.class); } }
@Test public void shouldThrowWhenConvertRawByteArrayFromConnectData() { try { converter.fromConnectData(TOPIC, Schema.BOOLEAN_SCHEMA, null); fail("now expected exception thrown"); } catch (Exception e) { assertThat(e).isExactlyInstanceOf(DataException.class); } }
@Test public void shouldThrowWhenNoDelegateConverterConfigured() { try { converter.fromConnectData(TOPIC, Schema.OPTIONAL_STRING_SCHEMA, "Hello World"); fail("now expected exception thrown"); } catch (Exception e) { assertThat(e).isExactlyInstanceOf(DataException.class); } }