KafkaQueue (ddth-queue 0.3.3 API)

java.lang.Object
- com.github.ddth.queue.impl.KafkaQueue

All Implemented Interfaces:

IQueue

Direct Known Subclasses:

BaseUniversalKafkaQueue
```
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()

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.
`int`	`getBufferSize()` Number of maximum items `KafkaQueue` should buffer.
`String`	`getConsumerGroupId()` Kafka's group-id to consume messages.
`protected com.github.ddth.kafka.KafkaClient`	`getKafkaClient()`
`long`	`getOffsetCommitPeriodMs()` Consumer's topic offset will be committed to server every `offsetCommitPeriodMs` milliseconds.Number of maximum items `KafkaQueue` should buffer.
`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()` Name of Kafka topic to store queue messages.
`String`	`getZkConnString()` Zookeeper connection string (format `host1:port1,host2:port2,host3:port3/path`).
`KafkaQueue`	`init()` Init method.
`boolean`	`isLeaderAutoRebalance()` Should we allow leaders to rebalance? Default value is `false`.
`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`	`setBufferSize(int bufferSize)`
`KafkaQueue`	`setConsumerGroupId(String consumerGroupId)`
`KafkaQueue`	`setKafkaClient(com.github.ddth.kafka.KafkaClient kafkaClient)` An external `KafkaClient` can be used.
`KafkaQueue`	`setLeaderAutoRebalance(boolean leaderAutoRebalance)`
`KafkaQueue`	`setOffsetCommitPeriodMs(long offsetCommitPeriodMs)`
`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.
`protected IQueueMessage`	`takeFromQueue()` Takes a message from Kafka 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()
```
    Name of Kafka topic to store queue messages.
    
    Returns:
  - setTopicName
```
public KafkaQueue setTopicName(String topicName)
```
  - isLeaderAutoRebalance
```
public boolean isLeaderAutoRebalance()
```
    Should we allow leaders to rebalance? Default value is false.
    
    Returns:
  - setLeaderAutoRebalance
```
public KafkaQueue setLeaderAutoRebalance(boolean leaderAutoRebalance)
```
  - getConsumerGroupId
```
public String getConsumerGroupId()
```
    Kafka's group-id to consume messages.
    
    Returns:
  - setConsumerGroupId
```
public KafkaQueue setConsumerGroupId(String consumerGroupId)
```
  - getBufferSize
```
public int getBufferSize()
```
    Number of maximum items KafkaQueue should buffer. Default value is 1. Higher value could gain more performance in consuming messages, but could cause message lost if client crashes.
    
    Returns:
  - setBufferSize
```
public KafkaQueue setBufferSize(int bufferSize)
```
  - getOffsetCommitPeriodMs
```
public long getOffsetCommitPeriodMs()
```
    Consumer's topic offset will be committed to server every offsetCommitPeriodMs milliseconds.Number of maximum items KafkaQueue should buffer. Default value is 1. Higher value could gain more performance in consuming messages, but could cause message lost if client crashes.
    
    Returns:
  - setOffsetCommitPeriodMs
```
public KafkaQueue setOffsetCommitPeriodMs(long offsetCommitPeriodMs)
```
  - getKafkaClient
```
protected com.github.ddth.kafka.KafkaClient getKafkaClient()
```
  - setKafkaClient
```
public KafkaQueue setKafkaClient(com.github.ddth.kafka.KafkaClient kafkaClient)
```
    An external KafkaClient can be used. If not set, KafkaClient will automatically create a KafkaClient for its own use.
    
    Parameters:
    kafkaClient -
    
    Returns:
  - init
```
public KafkaQueue init()
                throws Exception
```
    Init method.
    
    Returns:
    
    Throws:
    
    Exception
  - destroy
```
public void destroy()
```
    Destroy method.
  - serialize
```
protected abstract byte[] serialize(IQueueMessage msg)
                             throws QueueException
```
    Serializes a queue message to store in Kafka.
    
    Parameters:
    msg -
    
    Returns:
    
    Throws:
    
    QueueException
  - deserialize
```
protected abstract IQueueMessage deserialize(byte[] msgData)
                                      throws QueueException
```
    Deserilizes a queue message.
    
    Parameters:
    msgData -
    
    Returns:
    
    Throws:
    
    QueueException
  - takeFromQueue
```
protected IQueueMessage takeFromQueue()
```
    Takes a message from Kafka queue.
    
    Returns:
    Since:
    
    0.3.3
  - 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:

Class KafkaQueue

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Constructor Detail

KafkaQueue

Method Detail

getProducerType

setProducerType

getZkConnString

setZkConnString

getTopicName

setTopicName

isLeaderAutoRebalance

setLeaderAutoRebalance

getConsumerGroupId

setConsumerGroupId

getBufferSize

setBufferSize

getOffsetCommitPeriodMs

setOffsetCommitPeriodMs

getKafkaClient

setKafkaClient

init

destroy

serialize

deserialize

takeFromQueue

putToQueue

queue

requeue

requeueSilent

finish

take

getOrphanMessages

moveFromEphemeralToQueueStorage

queueSize

ephemeralSize