@Override
  public long getSplitBacklogBytes() {
    return backlogReader.computeMessageStats(fetchOffset).getMessageBytes();
  }
@Test
  public void getSplitBacklogBytes() throws Exception {
    startSubscriber();
    advancePastMessage(2);
    doReturn(ComputeMessageStatsResponse.newBuilder().setMessageBytes(42).build())
        .when(backlogReader)
        .computeMessageStats(Offset.of(3));
    assertEquals(42, reader.getSplitBacklogBytes());
  }