Page Comparison

MirrorMaker is not much more than a Kafka client that consumes from source topics and writes the same messages to some destination. As such, MirrorMaker does not offer any guarantees about a message from the source being written to the same partition and offset in the destination. The log saver and metrics processor store Kafka offsets per topic partition in HBase, and will need to be modified to handle a cluster failover. 

CDAP System Services

To start, we will require that the 'logs.user-v2' and 'metrics' topics have the same number of partitions in both clusters. This will guarantee that a message written to partition X in the master will also be written to the same partition X in the slave. Adding partitions to a topic will need to be handled specially.

...

Log saver only needs to be changed so that on start up, if the offset does not match the event time, it does a binary search for the correct offset. This is possible because log messages in Kafka contain the timestamp of the log message.

Metrics Processor

The metrics processor service stores its offsets in the 'metrics.kafka.meta' table. Row key is topic.partition, and the offset is stored under column 'o'. The logic is handled in KafkaConsumerMetaTable.

...

Metrics processor needs to be changed to also store the timestamp for each offset. On startup, if the offset and timestamp don't match, it does a binary search for the correct offset for that timestamp. This is possible because metrics messages in Kafka contain the timestamp of the metric.

Edge Conditions

If there are multiple Kafka messages for the same timestamp, some logs and metrics for that timestamp may be duplicated in the slave. 

Kafka Flowlet library

There is a library that can be used to write a flowlet that reads from Kafka (https://github.com/caskdata/cdap-packs/tree/develop/cdap-kafka-pack).

If a developer is using that library, it is up to the developer to handle the scenario when an offset may be pointing to different data after a failover. This is because the logic really requires application specific knowledge. For example, some applications may not be able to self correct if the Kafka messages don't contain some sort of monotonically increasing field, some applications may decide it is ok to re-process Kafka messages, and some may require some manual intervention to set the new offset through runtime arguments.

In any case, the developer has to override the method in the flowlet that gets the start offset, and probably also the method that saves the read offsets. For example, it could look like:

  @Override
  protected void saveReadOffsets(Map<TopicPartition, Long> offsets) {
    KeyValueTable offsetStore = getOffsetStore();
    if (offsetStore == null) {
      return;
    }

    for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
      TopicPartition topicPartition = entry.getKey();
      Long kafkaOffset = entry.getValue();
      // get the message id at this offset (assumes messages in Kafka have a monotonically increasing id)
      long messageId = getMessageId(topicPartition, kafkaOffset);
      // store both offset and message id
      offsetStore.write(getStoreKey(topicPartition), Bytes.add(Bytes.toBytes(kafkaOffset), Bytes.toBytes(messageId)));
    }
  }
 
  @Override
  protected Long getBeginOffset(TopicPartition topicPartition) {
    KeyValueTable offsetStore = getOffsetStore();

    if (offsetStore == null) {
      return getDefaultOffset(topicPartition);
    }

    // The value is simply a 8-bytes long representing the offset
    byte[] value = offsetStore.read(getStoreKey(topicPartition));
    if (value == null || value.length != Bytes.SIZEOF_LONG) {
      return getDefaultOffset(topicPartition);
    }
 
    long storedKafkaOffset = Bytes.toLong(value);
    long storedMessageId = Bytes.toLong(value, 8);
    
    // can happen after a cluster failover
    if (getMessageId(topicPartition, storedKafkaOffset) != storedMessageId) {
      return findKafkaOffset(storedMessageId);
    }
 
    return storedKafkaOffset;
  }

In this example, both the kafka offset and some message id corresponding to that offset are stored in the offset table. On program start, the message id from the offset store is compared to the actual message id in Kafka, and corrections are made if they do not match. This contrasts with the base class implementation (https://github.com/caskdata/cdap-packs/blob/develop/cdap-kafka-pack/cdap-kafka-flow/cdap-kafka-flow-compat-0.8/src/main/java/co/cask/cdap/kafka/flow/Kafka08ConsumerFlowlet.java), that just stores the kafka offset in the offset table and always assumes that it is correct.