LeaderData.clients has leaking possibility

[xirc]

Situation

1. A leader received a Replicate message from an entity and will be replicating an entry of the message for the entity.
   • The leader register this replication (ClientContext with LogEntryIndex) into LeaderData.clients.
2. The leader becomes a follower for some reason before it completes the replication.
   • There is another leader at this point.
3. The new leader completes the replication.
4. The old leader (a follower now) sends Replica to the entity.
   • The entry (ClientContext with LogEntryIndex) of LeaderData.clients is not removed.

Related source code

• akka-entity-replication/src/main/scala/lerna/akka/entityreplication/raft/RaftActor.scala

  Lines 240 to 265 in abe3af9

  	case FollowedLeaderCommit(leaderMember, leaderCommit) =>
  	currentData
  	.detectLeaderMember(leaderMember)
  	.followLeaderCommit(leaderCommit)
  	.applyCommittedLogEntries { logEntries =>
  	logEntries.foreach { logEntry =>
  	applyToReplicationActor(logEntry)
  	}
  	}
  	case Committed(logEntryIndex) =>
  	currentData
  	.commit(logEntryIndex)
  	.handleCommittedLogEntriesAndClients { entries =>
  	entries.foreach {
  	case (logEntry, Some(client)) =>
  	if (log.isDebugEnabled)
  	log.debug("=== [Leader] committed {} and will notify it to {} ===", logEntry, client)
  	client.ref.tell(
  	ReplicationSucceeded(logEntry.event.event, logEntry.index, client.instanceId),
  	client.originSender.getOrElse(ActorRef.noSender),
  	)
  	case (logEntry, None) =>
  	// 復旧中の commit or リーダー昇格時に未コミットのログがあった場合の commit
  	applyToReplicationActor(logEntry)
  	}
  	}

• akka-entity-replication/src/main/scala/lerna/akka/entityreplication/raft/RaftMemberData.scala

  Lines 226 to 231 in abe3af9

  	def handleCommittedLogEntriesAndClients(handler: Seq[(LogEntry, Option[ClientContext])] => Unit): RaftMemberData = {
  	val applicableLogEntries = selectApplicableLogEntries
  	handler(applicableLogEntries.map(e => (e, clients.get(e.index))))
  	updateVolatileState(lastApplied = applicableLogEntries.lastOption.map(_.index).getOrElse(lastApplied))
  	.updateLeaderVolatileState(clients = clients -- applicableLogEntries.map(_.index)) // 通知したクライアントは削除してメモリを節約
  	}