@Override
  public BytesStreamMessage getStreamMessage(int index) {
    return _messages.get(index);
  }
@Test
  public void testMessageBatchWithStitching() {
    PulsarConfig config = mock(PulsarConfig.class);
    when(config.getEnableKeyValueStitch()).thenReturn(true);
    List<BytesStreamMessage> streamMessages = List.of(PulsarUtils.buildPulsarStreamMessage(_message, config));
    PulsarMessageBatch messageBatch = new PulsarMessageBatch(streamMessages, mock(MessageIdStreamOffset.class), false);
    BytesStreamMessage streamMessage = messageBatch.getStreamMessage(0);
    byte[] keyValueBytes = streamMessage.getValue();
    assertNotNull(keyValueBytes);
    assertEquals(keyValueBytes.length, 8 + _expectedKeyBytes.length + _expectedValueBytes.length);
    ByteBuffer byteBuffer = ByteBuffer.wrap(keyValueBytes);
    int keyLength = byteBuffer.getInt();
    byte[] keyBytes = new byte[keyLength];
    byteBuffer.get(keyBytes);
    assertEquals(keyBytes, _expectedKeyBytes);
    int valueLength = byteBuffer.getInt();
    byte[] valueBytes = new byte[valueLength];
    byteBuffer.get(valueBytes);
    assertEquals(valueBytes, _expectedValueBytes);
  }