CassandraPersistenceQueries should find the highest sequence number even if deleted partition exists

[xirc]

For #201

CassandraPersistenceQueries doesn't yet support deleted partitions (which are empty). CassandraPersistenceQueries can handle one empty partition but not two or more consecutive empty partitions:

akka-entity-replication/rollback-tool-cassandra/src/main/scala/lerna/akka/entityreplication/rollback/cassandra/CassandraPersistenceQueries.scala

Lines 59 to 93 in 1afe799

	/** Finds the highest sequence number from the given partition or above
	*
	* If there are no events whose partition numbers are greater than or equal to the given partition, this method
	* returns `Future.successful(None)`.
	*
	* Since `akka.persistence.AtomicWrite` can skip at most one entire partition, the partition gap (an empty partition
	* exists between non-empty partitions) could occur. This method can handle such a partition gap.
	*
	* @see [[findHighestPartitionNr]]
	*/
	def findHighestSequenceNr(
	persistenceId: String,
	from: PartitionNr,
	): Future[Option[SequenceNr]] = {
	val allowableConsecutiveEmptyPartitionCount: Int = 1
	def find(
	currentPartitionNr: PartitionNr,
	lastHighestSequenceNr: Option[SequenceNr],
	consecutiveEmptyPartitionCount: Int,
	): Future[Option[SequenceNr]] = {
	selectHighestSequenceNr(persistenceId, currentPartitionNr)
	.flatMap {
	case Some(highestSequenceNr) =>
	assert(highestSequenceNr.value > 0)
	find(currentPartitionNr + 1, Some(highestSequenceNr), 0)
	case None =>
	if (consecutiveEmptyPartitionCount < allowableConsecutiveEmptyPartitionCount) {
	find(currentPartitionNr + 1, lastHighestSequenceNr, consecutiveEmptyPartitionCount + 1)
	} else {
	Future.successful(lastHighestSequenceNr)
	}
	}
	}
	find(from, None, 0)
	}

For example, CassandraPersistenceQueries.findHighestSequenceNr returns a Future containing None mistakenly in some cases if two or more consecutive empty partitions exist (all events on those partitions have been deleted).

Suppose that:

• Partition 0 (sequence numbers: 1 ~ 10): This partition is empty. All events on this partition have been deleted.
• Partition 1 (sequence numbers: 11 ~ 20): This partition is empty. All events on this partition have been deleted.
• Partition 2 (sequence numbers: 21 ~ 30): Events with sequence numbers from 24 up to 26 exist. Events with sequence numbers 21 up to 23 have been deleted.

CassandraPersistenceQueries.findHighestSequenceNr(persistenceId=???,from=PartitionNr(0)) returns a Future containing None now but should return the highest sequence number 26 instead.

Because LinearSequenceNrSearchStrategy (https://github.com/lerna-stack/akka-entity-replication/blob/v2.2.0/rollback-tool-cassandra/src/main/scala/lerna/akka/entityreplication/rollback/LinearSequenceNrSearchStrategy.scala) depends on CassandraPersistenceQueries.findHighestSequenceNr indirectly, a rollback might be impossible if two or more deleted partitions exist.

To address this issue, CassandraPersistenceQueries can fetch the highest deleted sequence number (called deleted_to) from the metadata table (akka.metadata) and then skip deleted partitions.

Akka Persistence Cassandra Journal Schema is described as:

• https://doc.akka.io/docs/akka-persistence-cassandra/1.0.5/journal.html#schema
• https://github.com/akka/akka-persistence-cassandra/blob/v1.0.5/core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournal.scala#L526-L628