@Override
  @SuppressWarnings("assignment")
  public ComputeMessageStatsResponse computeMessageStats(Offset offset) throws ApiException {
    try {
      return client
          .computeMessageStats(topicPath, partition, offset, Offset.of(Long.MAX_VALUE))
          .get(1, MINUTES);
    } catch (Throwable t) {
      throw toCanonical(t).underlying;
    }
  }
@Test
  public void computeMessageStats_validResponseCached() {
    Timestamp minEventTime = Timestamp.newBuilder().setSeconds(1000).setNanos(10).build();
    Timestamp minPublishTime = Timestamp.newBuilder().setSeconds(1001).setNanos(11).build();
    ComputeMessageStatsResponse response =
        ComputeMessageStatsResponse.newBuilder()
            .setMessageCount(10)
            .setMessageBytes(100)
            .setMinimumEventTime(minEventTime.toBuilder().setSeconds(1002).build())
            .setMinimumPublishTime(minPublishTime)
            .build();

    when(mockClient.computeMessageStats(
            example(TopicPath.class),
            example(Partition.class),
            example(Offset.class),
            Offset.of(Long.MAX_VALUE)))
        .thenReturn(ApiFutures.immediateFuture(response));

    assertEquals(reader.computeMessageStats(example(Offset.class)), response);
  }