com.github.ddth.queue.impl

Class KafkaQueue

java.lang.Object

com.github.ddth.queue.impl.KafkaQueue

All Implemented Interfaces:

IQueue

Direct Known Subclasses:

UniversalKafkaQueue

public abstract class KafkaQueue
extends Object
implements IQueue

(Experimental) Kafka implementation of IQueue.

Since:

0.3.2

Author:

Thanh Ba Nguyen

Constructor Summary

Constructors

Constructor and Description
KafkaQueue()

Method Summary

Methods

Modifier and Type	Method and Description
protected abstract IQueueMessage	deserialize(byte[] msgData) Deserilizes a queue message.
void	destroy() Destroy method.
int	ephemeralSize() Gets ephemeral-storage's number of items.
void	finish(IQueueMessage msg) Called when finish processing the message to cleanup ephemeral storage.
protected com.github.ddth.kafka.KafkaClient	getKafkaClient()
Collection<IQueueMessage>	getOrphanMessages(long thresholdTimestampMs) Gets all orphan messages (messages that were left in ephemeral storage for a long time).
com.github.ddth.kafka.KafkaClient.ProducerType	getProducerType()
String	getTopicName()
String	getZkConnString() Zookeeper connection string (format host1:port1,host2:port2,host3:port3/path).
KafkaQueue	init() Init method.
boolean	moveFromEphemeralToQueueStorage(IQueueMessage msg) Moves a message from ephemeral back to queue storage.
protected boolean	putToQueue(IQueueMessage msg) Puts a message to Kafka queue.
boolean	queue(IQueueMessage msg) Queues a message.
int	queueSize() Gets queue's number of items.
boolean	requeue(IQueueMessage msg) Re-queues a message.
boolean	requeueSilent(IQueueMessage msg) Silently re-queues a message.
protected abstract byte[]	serialize(IQueueMessage msg) Serializes a queue message to store in Kafka.
KafkaQueue	setKafkaClient(com.github.ddth.kafka.KafkaClient kafkaClient)
KafkaQueue	setProducerType(com.github.ddth.kafka.KafkaClient.ProducerType producerType)
KafkaQueue	setTopicName(String topicName)
KafkaQueue	setZkConnString(String zkConnString) Sets Zookeeper connection string (format host1:port1,host2:port2,host3:port3/path).
IQueueMessage	take() Takes a message out of queue.

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaQueue

public KafkaQueue()

Method Detail

getProducerType

public com.github.ddth.kafka.KafkaClient.ProducerType getProducerType()

setProducerType

public KafkaQueue setProducerType(com.github.ddth.kafka.KafkaClient.ProducerType producerType)

getZkConnString

public String getZkConnString()

Zookeeper connection string (format host1:port1,host2:port2,host3:port3/path).

Returns:

setZkConnString

public KafkaQueue setZkConnString(String zkConnString)

Sets Zookeeper connection string (format host1:port1,host2:port2,host3:port3/path).

Parameters:

redisHostAndPort -

Returns:

getTopicName

public String getTopicName()

setTopicName

public KafkaQueue setTopicName(String topicName)

getKafkaClient

protected com.github.ddth.kafka.KafkaClient getKafkaClient()

setKafkaClient

public KafkaQueue setKafkaClient(com.github.ddth.kafka.KafkaClient kafkaClient)

init

public KafkaQueue init()
                throws Exception

Init method.

Returns:

Throws:

Exception

destroy

public void destroy()

Destroy method.

serialize

protected abstract byte[] serialize(IQueueMessage msg)

Serializes a queue message to store in Kafka.

Parameters:

msg -

Returns:

deserialize

protected abstract IQueueMessage deserialize(byte[] msgData)

Deserilizes a queue message.

Parameters:

msgData -

Returns:

putToQueue

protected boolean putToQueue(IQueueMessage msg)

Puts a message to Kafka queue.

Parameters:

msg -

Returns:

queue

public boolean queue(IQueueMessage msg)

Queues a message.

Implementation flow:

• Put message to tail of queue storage.

Specified by:

queue in interface IQueue

Returns:

requeue

public boolean requeue(IQueueMessage msg)

Re-queues a message.

Implementation flow:

• Put message to tail of queue storage; and increase message's re-queue count & update message's queue timestamp.
• Remove message from ephemeral storage.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

requeue in interface IQueue

Returns:

requeueSilent

public boolean requeueSilent(IQueueMessage msg)

Silently re-queues a message.

Implementation flow:

• Put message to tail of queue storage; do NOT increase message's re-queue count and do NOT update message's queue timestamp.
• Remove message from ephemeral storage.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

requeueSilent in interface IQueue

Returns:

finish

public void finish(IQueueMessage msg)

Called when finish processing the message to cleanup ephemeral storage.

Implementation flow:

• Remove message from ephemeral storage.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

finish in interface IQueue

take

public IQueueMessage take()

Takes a message out of queue.

Implementation flow:

• Read message from head of queue storage.
• Write message to ephemeral storage.
• Remove message from queue storage.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

take in interface IQueue

Returns:

getOrphanMessages

public Collection<IQueueMessage> getOrphanMessages(long thresholdTimestampMs)

Gets all orphan messages (messages that were left in ephemeral storage for a long time).

Specified by:

getOrphanMessages in interface IQueue

Parameters:

thresholdTimestampMs -

Returns:

moveFromEphemeralToQueueStorage

public boolean moveFromEphemeralToQueueStorage(IQueueMessage msg)

Moves a message from ephemeral back to queue storage. Useful when dealing with orphan messages.

Implementation flow:

• Read message from the ephemeral storage.
• Put the message back to queue.
• Remove the message from the ephemeral storage.

Specified by:

moveFromEphemeralToQueueStorage in interface IQueue

Returns:

true if a move has been made, false otherwise (e.g. the message didn't exist in ephemeral storage)

queueSize

public int queueSize()

Gets queue's number of items.

Specified by:

queueSize in interface IQueue

Returns:

ephemeralSize

public int ephemeralSize()

Gets ephemeral-storage's number of items.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

ephemeralSize in interface IQueue

Returns: