[ISSUE #134] fix batch push error (#137)

Signed-off-by: zhangyang <Git_Yang@163.com>

Author: Git-Yang

src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java

@@ -589,7 +589,7 @@ private void doAppend() throws Exception {
         private void sendBatchAppendEntryRequest() throws Exception {
             batchAppendEntryRequest.setCommitIndex(dLedgerStore.getCommittedIndex());
             CompletableFuture<PushEntryResponse> responseFuture = dLedgerRpcService.push(batchAppendEntryRequest);
-            batchPendingMap.put(batchAppendEntryRequest.getFirstEntryIndex(), new Pair<>(System.currentTimeMillis(), batchAppendEntryRequest.getCount()));
+            batchPendingMap.put(batchAppendEntryRequest.getLastEntryIndex(), new Pair<>(System.currentTimeMillis(), batchAppendEntryRequest.getCount()));
             responseFuture.whenComplete((x, ex) -> {
                 try {
                     PreConditions.check(ex == null, DLedgerResponseCode.UNKNOWN);
@@ -890,24 +890,13 @@ private PushEntryResponse buildResponse(PushEntryRequest request, int code) {
             response.setCode(code);
             response.setTerm(request.getTerm());
             if (request.getType() != PushEntryRequest.Type.COMMIT) {
-                response.setIndex(request.getEntry().getIndex());
+                response.setIndex(request.getLastEntryIndex());
             }
             response.setBeginIndex(dLedgerStore.getLedgerBeginIndex());
             response.setEndIndex(dLedgerStore.getLedgerEndIndex());
             return response;
         }
 
-        private PushEntryResponse buildBatchAppendResponse(PushEntryRequest request, int code) {
-            PushEntryResponse response = new PushEntryResponse();
-            response.setGroup(request.getGroup());
-            response.setCode(code);
-            response.setTerm(request.getTerm());
-            response.setIndex(request.getLastEntryIndex());
-            response.setBeginIndex(dLedgerStore.getLedgerBeginIndex());
-            response.setEndIndex(dLedgerStore.getLedgerEndIndex());
-            return response;
-        }
-
         private void handleDoAppend(long writeIndex, PushEntryRequest request,
             CompletableFuture<PushEntryResponse> future) {
             try {
@@ -975,7 +964,7 @@ private void handleDoBatchAppend(long writeIndex, PushEntryRequest request,
                 for (DLedgerEntry entry : request.getBatchEntry()) {
                     dLedgerStore.appendAsFollower(entry, request.getTerm(), request.getLeaderId());
                 }
-                future.complete(buildBatchAppendResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
+                future.complete(buildResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
                 updateCommittedIndex(request.getTerm(), request.getCommitIndex());
             } catch (Throwable t) {
                 logger.error("[HandleDoBatchAppend]", t);
@@ -999,11 +988,11 @@ private void checkAppendFuture(long endIndex) {
                             DLedgerEntry dLedgerEntry = pair.getKey().getEntry();
                             PreConditions.check(dLedgerEntry.equals(dLedgerStore.get(dLedgerEntry.getIndex())), DLedgerResponseCode.INCONSISTENT_STATE);
                         }
-                        pair.getValue().complete(buildBatchAppendResponse(pair.getKey(), DLedgerResponseCode.SUCCESS.getCode()));
+                        pair.getValue().complete(buildResponse(pair.getKey(), DLedgerResponseCode.SUCCESS.getCode()));
                         logger.warn("[PushFallBehind]The leader pushed an batch append entry last index={} smaller than current ledgerEndIndex={}, maybe the last ack is missed", lastEntryIndex, endIndex);
                     } catch (Throwable t) {
                         logger.error("[PushFallBehind]The leader pushed an batch append entry last index={} smaller than current ledgerEndIndex={}, maybe the last ack is missed", lastEntryIndex, endIndex, t);
-                        pair.getValue().complete(buildBatchAppendResponse(pair.getKey(), DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
+                        pair.getValue().complete(buildResponse(pair.getKey(), DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
                     }
                     writeRequestMap.remove(pair.getKey().getFirstEntryIndex());
                     continue;
@@ -1027,7 +1016,7 @@ private void checkAppendFuture(long endIndex) {
                 return;
             }
             logger.warn("[PushFastForward] ledgerEndIndex={} entryIndex={}", endIndex, minFastForwardIndex);
-            pair.getValue().complete(buildBatchAppendResponse(pair.getKey(), DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
+            pair.getValue().complete(buildResponse(pair.getKey(), DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
         }
         /**
          * The leader does push entries to follower, and record the pushed index. But in the following conditions, the push may get stopped.