import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.japi.Creator;
+import com.google.common.base.Stopwatch;
+import java.util.concurrent.TimeUnit;
+import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
private final Configuration configuration;
private final String followerId;
private final Logger LOG = LoggerFactory.getLogger(DummyShard.class);
+ private long lastMessageIndex = -1;
+ private long lastMessageSize = 0;
+ private Stopwatch appendEntriesWatch;
public DummyShard(Configuration configuration, String followerId) {
this.configuration = configuration;
}
protected void handleAppendEntries(AppendEntries req) throws InterruptedException {
- LOG.info("{} - Received AppendEntries message : leader term, index, size = {}, {}, {}", followerId, req.getTerm(),req.getLeaderCommit(), req.getEntries().size());
+ LOG.info("{} - Received AppendEntries message : leader term = {}, index = {}, prevLogIndex = {}, size = {}",
+ followerId, req.getTerm(),req.getLeaderCommit(), req.getPrevLogIndex(), req.getEntries().size());
+
+ if(appendEntriesWatch != null){
+ long elapsed = appendEntriesWatch.elapsed(TimeUnit.SECONDS);
+ if(elapsed >= 5){
+ LOG.error("More than 5 seconds since last append entry, elapsed Time = {} seconds" +
+ ", leaderCommit = {}, prevLogIndex = {}, size = {}",
+ elapsed, req.getLeaderCommit(), req.getPrevLogIndex(), req.getEntries().size());
+ }
+ appendEntriesWatch.reset().start();
+ } else {
+ appendEntriesWatch = Stopwatch.createStarted();
+ }
+
+ if(lastMessageIndex == req.getLeaderCommit() && req.getEntries().size() > 0 && lastMessageSize > 0){
+ LOG.error("{} - Duplicate message with leaderCommit = {} prevLogIndex = {} received", followerId, req.getLeaderCommit(), req.getPrevLogIndex());
+ }
+
+ lastMessageIndex = req.getLeaderCommit();
+ lastMessageSize = req.getEntries().size();
+
long lastIndex = req.getLeaderCommit();
- if (req.getEntries().size() > 0)
- lastIndex = req.getEntries().get(0).getIndex();
+ if (req.getEntries().size() > 0) {
+ for(ReplicatedLogEntry entry : req.getEntries()) {
+ lastIndex = entry.getIndex();
+ }
+ }
- if (configuration.shouldCauseTrouble()) {
+ if (configuration.shouldCauseTrouble() && req.getEntries().size() > 0) {
boolean ignore = false;
if (configuration.shouldDropReplies()) {