org.thales.punch.apps.shiva.kafka

Class KafkaLeaderController

java.lang.Object

org.thales.punch.apps.shiva.kafka.KafkaLeaderController

All Implemented Interfaces:

Runnable, org.thales.punch.kafka.api.IRecordHandler<String,byte[]>

public class KafkaLeaderController
extends Object
implements Runnable, org.thales.punch.kafka.api.IRecordHandler<String,byte[]>

This class takes care of the kafka gossip exchange required by the leader.

Here is the logic :

• Workers registration and discovery
  • The leader consumes the data topic using the last committed strategy.
  • Only one leader in the shiva cluster obtains the leadership on that data topic.
  • At startup if the leader is late it will replay all the start/stop commands submitted while no leader was there.
    • This replay does not pose problem because of the way the assignment is periodically refreshed by workers. I.e. they will not be a storm of start and stop commands executed on the cluster.
• Start and stop commands handling
  • The leader consumes the control topic using the last committed strategy.
  • Only one leader in the shiva cluster obtains the leadership on that control topic.
  • At startup if the leader is late it will replay all the start/stop commands submitted while no leader was there.
    • This replay does not pose problem because of the way the assignment is periodically refreshed by workers. I.e. they will not be a storm of start and stop commands executed on the cluster.

Author:

dimi

Constructor Summary

Constructors

Constructor and Description
KafkaLeaderController(String bootstrapServers, org.apache.kafka.clients.producer.KafkaProducer<String,byte[]> producer, String clusterName, String commandTopic, String assignementTopic, String controlTopic, ILeader leader, long tickInterval, org.thales.punch.platform.api.pojo.JavaStoreSecurity securityStore)

Method Summary

Modifier and Type	Method and Description
void	onPartitionAssigned(org.thales.punch.kafka.api.IPartition partition, org.apache.kafka.clients.consumer.OffsetAndMetadata offsetAndMetadata)
void	onPartitionRevoked(org.thales.punch.kafka.api.IPartition partition)
void	onReceive(org.thales.punch.kafka.api.IPartition partition, org.apache.kafka.clients.consumer.ConsumerRecord<String,byte[]> record)
void	onTick(org.thales.punch.kafka.api.IPartition partition, long nextOffsetToRead)
void	run()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

KafkaLeaderController

public KafkaLeaderController(String bootstrapServers,
                             org.apache.kafka.clients.producer.KafkaProducer<String,byte[]> producer,
                             String clusterName,
                             String commandTopic,
                             String assignementTopic,
                             String controlTopic,
                             ILeader leader,
                             long tickInterval,
                             org.thales.punch.platform.api.pojo.JavaStoreSecurity securityStore)

Method Detail

onReceive

public void onReceive(org.thales.punch.kafka.api.IPartition partition,
                      org.apache.kafka.clients.consumer.ConsumerRecord<String,byte[]> record)

Specified by:

onReceive in interface org.thales.punch.kafka.api.IRecordHandler<String,byte[]>

onPartitionAssigned

public void onPartitionAssigned(org.thales.punch.kafka.api.IPartition partition,
                                org.apache.kafka.clients.consumer.OffsetAndMetadata offsetAndMetadata)

Specified by:

onPartitionAssigned in interface org.thales.punch.kafka.api.IRecordHandler<String,byte[]>

onPartitionRevoked

public void onPartitionRevoked(org.thales.punch.kafka.api.IPartition partition)

Specified by:

onPartitionRevoked in interface org.thales.punch.kafka.api.IRecordHandler<String,byte[]>

run

public void run()

Specified by:

run in interface Runnable

onTick

public void onTick(org.thales.punch.kafka.api.IPartition partition,
                   long nextOffsetToRead)

Specified by:

onTick in interface org.thales.punch.kafka.api.IRecordHandler<String,byte[]>