@Override
  public StreamDataDecoderResult decode(StreamMessage message) {
    assert message.getValue() != null;

    try {
      _reuse.clear();
      GenericRow row = _valueDecoder.decode(message.getValue(), 0, message.getLength(), _reuse);
      if (row != null) {
        if (message.getKey() != null) {
          row.putValue(KEY, new String(message.getKey(), StandardCharsets.UTF_8));
        }
        StreamMessageMetadata metadata = message.getMetadata();
        if (metadata != null) {
          if (metadata.getHeaders() != null) {
            metadata.getHeaders().getFieldToValueMap()
                .forEach((key, value) -> row.putValue(HEADER_KEY_PREFIX + key, value));
          }
          if (metadata.getRecordMetadata() != null) {
            metadata.getRecordMetadata().forEach((key, value) -> row.putValue(METADATA_KEY_PREFIX + key, value));
          }
        }
        return new StreamDataDecoderResult(row, null);
      } else {
        return new StreamDataDecoderResult(null,
            new RuntimeException("Encountered unknown exception when decoding a Stream message"));
      }
    } catch (Exception e) {
      LOGGER.error("Failed to decode StreamMessage", e);
      return new StreamDataDecoderResult(null, e);
    }
  }
@Test
  public void testNoExceptionIsThrown()
      throws Exception {
    ThrowingDecoder messageDecoder = new ThrowingDecoder();
    messageDecoder.init(Collections.emptyMap(), ImmutableSet.of(NAME_FIELD), "");
    String value = "Alice";
    BytesStreamMessage message = new BytesStreamMessage(value.getBytes(StandardCharsets.UTF_8));
    StreamDataDecoderResult result = new StreamDataDecoderImpl(messageDecoder).decode(message);
    Assert.assertNotNull(result);
    Assert.assertNotNull(result.getException());
    Assert.assertNull(result.getResult());
  }