@Override
  public int compare(
      Entry<AsyncLogger, PrepareRecoveryResponseProto> a,
      Entry<AsyncLogger, PrepareRecoveryResponseProto> b) {
    
    PrepareRecoveryResponseProto r1 = a.getValue();
    PrepareRecoveryResponseProto r2 = b.getValue();
    
    // A response that has data for a segment is always better than one
    // that doesn't.
    if (r1.hasSegmentState() != r2.hasSegmentState()) {
      return Booleans.compare(r1.hasSegmentState(), r2.hasSegmentState());
    }
    
    if (!r1.hasSegmentState()) {
      // Neither has a segment, so neither can be used for recover.
      // Call them equal.
      return 0;
    }
    
    // They both have a segment.
    SegmentStateProto r1Seg = r1.getSegmentState();
    SegmentStateProto r2Seg = r2.getSegmentState();
    
    Preconditions.checkArgument(r1Seg.getStartTxId() == r2Seg.getStartTxId(),
        "Should only be called with responses for corresponding segments: " +
        "%s and %s do not have the same start txid.", r1, r2);

    // If one is in-progress but the other is finalized,
    // the finalized one is greater.
    if (r1Seg.getIsInProgress() != r2Seg.getIsInProgress()) {
      return Booleans.compare(!r1Seg.getIsInProgress(), !r2Seg.getIsInProgress());
    }
    
    if (!r1Seg.getIsInProgress()) {
      // If both are finalized, they should match lengths
      if (r1Seg.getEndTxId() != r2Seg.getEndTxId()) {
        throw new AssertionError("finalized segs with different lengths: " + 
            r1 + ", " + r2);
      }
      return 0;
    }
    
    // Both are in-progress.
    long r1SeenEpoch = Math.max(r1.getAcceptedInEpoch(), r1.getLastWriterEpoch());
    long r2SeenEpoch = Math.max(r2.getAcceptedInEpoch(), r2.getLastWriterEpoch());
    
    return ComparisonChain.start()
        .compare(r1SeenEpoch, r2SeenEpoch)
        .compare(r1.getSegmentState().getEndTxId(), r2.getSegmentState().getEndTxId())
        .result();
  }
@Test
  public void testComparisons() {
    Entry<AsyncLogger, PrepareRecoveryResponseProto> INPROGRESS_1_3 =
        makeEntry(PrepareRecoveryResponseProto.newBuilder()
          .setSegmentState(SegmentStateProto.newBuilder()
              .setStartTxId(1L)
              .setEndTxId(3L)
              .setIsInProgress(true))
          .setLastWriterEpoch(0L)
          .build());
    Entry<AsyncLogger, PrepareRecoveryResponseProto> INPROGRESS_1_4 =
        makeEntry(PrepareRecoveryResponseProto.newBuilder()
          .setSegmentState(SegmentStateProto.newBuilder()
              .setStartTxId(1L)
              .setEndTxId(4L)
              .setIsInProgress(true))
          .setLastWriterEpoch(0L)
          .build());
    Entry<AsyncLogger, PrepareRecoveryResponseProto> INPROGRESS_1_4_ACCEPTED =
        makeEntry(PrepareRecoveryResponseProto.newBuilder()
          .setSegmentState(SegmentStateProto.newBuilder()
              .setStartTxId(1L)
              .setEndTxId(4L)
              .setIsInProgress(true))
          .setLastWriterEpoch(0L)
          .setAcceptedInEpoch(1L)
          .build());

    Entry<AsyncLogger, PrepareRecoveryResponseProto> FINALIZED_1_3 =
        makeEntry(PrepareRecoveryResponseProto.newBuilder()
          .setSegmentState(SegmentStateProto.newBuilder()
              .setStartTxId(1L)
              .setEndTxId(3L)
              .setIsInProgress(false))
          .setLastWriterEpoch(0L)
          .build());

    // Should compare equal to itself
    assertEquals(0, INSTANCE.compare(INPROGRESS_1_3, INPROGRESS_1_3));
    
    // Longer log wins.
    assertEquals(-1, INSTANCE.compare(INPROGRESS_1_3, INPROGRESS_1_4));
    assertEquals(1, INSTANCE.compare(INPROGRESS_1_4, INPROGRESS_1_3));
    
    // Finalized log wins even over a longer in-progress
    assertEquals(-1, INSTANCE.compare(INPROGRESS_1_4, FINALIZED_1_3));
    assertEquals(1, INSTANCE.compare(FINALIZED_1_3, INPROGRESS_1_4));

    // Finalized log wins even if the in-progress one has an accepted
    // recovery proposal.
    assertEquals(-1, INSTANCE.compare(INPROGRESS_1_4_ACCEPTED, FINALIZED_1_3));
    assertEquals(1, INSTANCE.compare(FINALIZED_1_3, INPROGRESS_1_4_ACCEPTED));
  }