You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 8 Next »

Status

Current state: Under discussion

Discussion thread: here

JIRA: KAFKA-1650, KAFKA-1839, KAFKA-1840

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

The Mirror Maker has a potential dataloss issue as explained below:

1. Mirror Maker consumer consume some messages and put them into dataChannel.

2. Mirror Maker consumer commits the offsets

3. Mirror Maker somehow dies with some messages left in the dataChannel not yet successfully delivered by producer.

4. When Mirror Maker restarts, all the messages that has been consumed and committed by consumer but not delivered by producer yet will be lost.

 

Currently Mirror Maker does not ensure the topics in source and target clusters have same partitions, which is required in many cases.

Sometimes it is useful for Mirror Maker to do some simple process on messages (filter, message format change, etc.)

Public Interfaces

The no data loss solution uses the callback of new producer to determine the offsets to be committed.

A ConsumerRebalanceListener class is created and could be wired into ZookeeperConsumerConnector to avoid duplicate messages when consumer rebalance occurs in mirror maker.

ConsumerRebalanceListener
/**
 * This listener is used for execution of tasks defined by user when a consumer rebalance
 * occurs in {@link kafka.consumer.ZookeeperConsumerConnector}
 */
public interface ConsumerRebalanceListener {
    /**
     * This method is called after all the fetcher threads are stopped but before the
     * ownership of partitions are released. Depending on whether auto offset commit is
     * enabled or not, offsets may or may not have been committed.
     * This listener is initially added to prevent duplicate messages on consumer rebalance
     * in mirror maker, where offset auto commit is disabled to prevent data loss. It could
     * also be used in more general cases.
     */
    public void beforeReleasingPartitions(Map<String, Set<Integer>> partitionOwnership);
    /**
     * This method is called after the new partition assignment is finished but before fetcher
     * threads start. A map of new global partition assignment is passed in as parameter.
     */
    public void beforeStartingFetchers(Map<String, Map<Integer, ConsumerThreadId>> partitionAssignment);
}

 

For Mirror Maker, an internal rebalance listener is wired in by default to avoid duplicates on consumer rebalance. User could still specify a custom listener class in command line argument. The internal rebalance listener will call that custom listener after it finishes the default logic.

Proposed Changes

The only source of truth for offsets in consume-then-send pattern is target cluster. The offsets should only be committed after having received the confirmation from the target cluster. For the new producer, the confirmation is the ACK from target cluster.

In that sense the consumer's offsets auto-commits should be turned off, and the offsets will be committed after a future associated with a message is received.

Maintaining message order (KAFKA-1650)

For both keyed and none-keyed messages, the TopicPartition's hash is used to decide which data channel queue they goes to. So the message order in data channel is the same as in source partitions.

Offset commit thread (KAFKA-1650)

The offsets commit thread should execute periodically. A Map[TopicPartition, UnackedOffsetList] will be needed to track the offsets that is ready for commit.

On consumer rebalance: (KAFKA-1650)

A new consumerRebalanceListener needs to be added to make sure there is no duplicates when consumer rebalance occurs. The callback beforeReleasingPartitions will be invoked  on consumer rebalance. It should

1. clear the dataChannel queue (as those messages will be re-consumed after rebalance)

2. wait until the producer drains all remaining messages

3. commit the offsets

Ensure same partition number in source and target topics (KAFKA-1839)

Introduce a customRebalanceListener with the callback beforeStartFetchers(newPartitionAssignmentMap). The callback is invoked when consumer rebalance occurs (topic creation/deletion, partition expansion). It will compare the new topic partition info with the existing one to see what change occurs. If a new topic is created, the callback create the topic in target cluster with the same number of partitions. This callback exists in new consumer (onPartitionsAssigned)

Message handler in consumer thread (KAFKA-1840)

Sometimes it is useful for mirror maker to do some process on the messages it consumed such as filtering out some messages or format conversions. This could be done by adding a message handler in consumer thread. The interface could be

public interface MirrorMakerMessageHandler {
    /**
     * The message handler will be executed to handle messages in mirror maker
     * consumer thread before the messages go into data channel. The message handler should
     * not block and should handle all the exceptions by itself.
     */
    public List<MirrorMakerRecord> handle(MirrorMakerRecord record);
}
The default message handler just return a list with passed in record.

Changes to make with new consumer

As of now new consumer is still under development. After new consumer is available, the following changes needs to be made:

  1. Change the message handler interface to take ConsumerRecord for both input and output.
  2. Change Consumer thread according to new consumer API.  The options are:
    1. Use a single consumer thread. (Preferred unless there is apparent performance issue)
    2. Use multiple consumer threads by decoupling consumption and processing.

Will submit follow up patch after the new consumer is available.

Compatibility, Deprecation, and Migration Plan

The changes are backward compatible.

Rejected Alternatives

None

  • No labels