@Override
  public BytesStreamMessage getStreamMessage(int index) {
    return _messages.get(index);
  }
@Test
  public void testMessageBatchNoStitching() {
    PulsarConfig config = mock(PulsarConfig.class);
    when(config.getEnableKeyValueStitch()).thenReturn(false);
    List<BytesStreamMessage> streamMessages = List.of(PulsarUtils.buildPulsarStreamMessage(_message, config));
    PulsarMessageBatch messageBatch = new PulsarMessageBatch(streamMessages, mock(MessageIdStreamOffset.class), false);
    byte[] valueBytes = messageBatch.getStreamMessage(0).getValue();
    assertEquals(valueBytes, _expectedValueBytes);
  }