Indexers should be tolerant of a Kafka broker offset rollback

[bryanlb]

Describe the bug

The indexer currently has logic that prevents running when the Kafka offset is rolled back. While this seemed like a good idea at the time, in the course of operating Kafka administrators can take actions that cause the offset to be reset. When this happens today, it results in an indexer becoming completely stuck, and can require deleting all of the data for that indexer (recovery tasks, and snapshots).

astra/astra/src/main/java/com/slack/astra/server/RecoveryTaskCreator.java

Lines 244 to 251 in 9838d38

	if (currentEndOffsetForPartition < highestDurableOffsetForPartition) {
	final String message =
	String.format(
	"The current head for the partition %d can't "
	+ "be lower than the highest durable offset for that partition %d",
	currentEndOffsetForPartition, highestDurableOffsetForPartition);
	LOG.error(message);
	throw new IllegalStateException(message);

We should rework this so rollbacks are supported. Additionally we may consider moving from calculating the offset from previous snapshots + pending recovery tasks to a separate ZK node that is only responsible for storing the offset for the indexer. This would allow easier resetting, and wouldn't require the indexer to load all of the snapshots into memory on boot (allowing quicker boot times).