class 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();
    }

}