public static BytesStreamMessage buildPulsarStreamMessage(Message<byte[]> message, PulsarConfig config) {
    byte[] key = message.getKeyBytes();
    byte[] value = message.getData();
    if (config.getEnableKeyValueStitch()) {
      value = stitchKeyValue(key, value);
    }
    return new BytesStreamMessage(key, value, extractMessageMetadata(message, config));
  }
@Test
  public void testPulsarSteamMessageUnstitched() {
    PulsarConfig config = mock(PulsarConfig.class);
    when(config.getEnableKeyValueStitch()).thenReturn(false);
    String key = "key";
    String value = "value";
    DummyPulsarMessage dummyPulsarMessage =
        new DummyPulsarMessage(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));
    BytesStreamMessage streamMessage = PulsarUtils.buildPulsarStreamMessage(dummyPulsarMessage, config);
    assertEquals(streamMessage.getKey(), key.getBytes(StandardCharsets.UTF_8));
    assertEquals(streamMessage.getValue(), value.getBytes(StandardCharsets.UTF_8));
  }