diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java index a6798c48b4..a3652f4976 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java @@ -1694,7 +1694,10 @@ private CompletableFuture appendLog(ReferenceCountedObject entriesTermIndices; try(UncheckedAutoCloseableSupplier> entries = entriesRef.retainAndReleaseOnClose()) { entriesTermIndices = ConsecutiveIndices.convert(entries.get()); - appendLogTermIndices.append(entriesTermIndices); + if (!appendLogTermIndices.append(entriesTermIndices)) { + // index already exists, return the last future + return appendLogFuture.get(); + } } entriesRef.retain(); diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java index c5010a5346..ce4702d958 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java @@ -136,15 +136,20 @@ synchronized Long getTerm(long index) { return floorEntry.getValue().getTerm(index); } - synchronized void append(List entriesTermIndices) { - for(ConsecutiveIndices indices : entriesTermIndices) { - // validate startIndex - final Map.Entry lastEntry = map.lastEntry(); - if (lastEntry != null) { - Preconditions.assertSame(lastEntry.getValue().getNextIndex(), indices.startIndex, "startIndex"); + synchronized boolean append(List entriesTermIndices) { + for(int i = 0; i < entriesTermIndices.size(); i++) { + final ConsecutiveIndices indices = entriesTermIndices.get(i); + final ConsecutiveIndices previous = map.put(indices.startIndex, indices); + if (previous != null) { + // index already exists, revert this append + map.put(previous.startIndex, previous); + for(int j = 0; j < i; j++) { + map.remove(entriesTermIndices.get(j).startIndex); + } + return false; } - map.put(indices.startIndex, indices); } + return true; } synchronized void removeExisting(List entriesTermIndices) {