KafkaQueue (ddth-queue-core 0.4.2 API)

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

All Implemented Interfaces:

IQueue, Closeable, AutoCloseable

Direct Known Subclasses:

BaseUniversalKafkaQueue
```
public abstract class KafkaQueue
extends Object
implements IQueue, Closeable, AutoCloseable
```
(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

All Methods Instance Methods Abstract Methods Concrete Methods
Modifier and Type	Method and Description
`void`	`close()`
`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.
`String`	`getConsumerGroupId()` Kafka's group-id to consume messages.
`String`	`getKafkaBootstrapServers()` Kafka bootstrap server list (format `host1:9092,host2:port2,host3:port3`).
`protected com.github.ddth.kafka.KafkaClient`	`getKafkaClient()`
`Properties`	`getKafkaConsumerProperties()` Gets Kafka consumer's custom configuration properties.
`Properties`	`getKafkaProducerProperties()` Gets Kafka producer's custom configuration properties.
`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.
`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, partitioning message by `IQueueMessage.qId()`
`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`	`setConsumerGroupId(String consumerGroupId)`
`KafkaQueue`	`setKafkaBootstrapServers(String kafkaBootstrapServers)` Sets Kafka bootstrap server list (format `host1:9092,host2:port2,host3:port3`).
`KafkaQueue`	`setKafkaClient(com.github.ddth.kafka.KafkaClient kafkaClient)` An external `KafkaClient` can be used.
`KafkaQueue`	`setKafkaConsumerProperties(Properties kafkaConsumerConfigs)` Sets Kafka consumer's custom configuration properties.
`KafkaQueue`	`setKafkaProducerProperties(Properties kafkaProducerConfigs)` Sets Kafka producer's custom configuration properties.
`KafkaQueue`	`setProducerType(com.github.ddth.kafka.KafkaClient.ProducerType producerType)`
`KafkaQueue`	`setTopicName(String topicName)`
`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)
```
  - getKafkaBootstrapServers
```
public String getKafkaBootstrapServers()
```
    Kafka bootstrap server list (format host1:9092,host2:port2,host3:port3).
    
    Returns:
    
    Since:
    
    0.4.0
  - setKafkaBootstrapServers
```
public KafkaQueue setKafkaBootstrapServers(String kafkaBootstrapServers)
```
    Sets Kafka bootstrap server list (format host1:9092,host2:port2,host3:port3).
    
    Parameters:
    
    kafkaBootstrapServers -
    
    Returns:
    
    Since:
    
    0.4.0
  - getKafkaProducerProperties
```
public Properties getKafkaProducerProperties()
```
    Gets Kafka producer's custom configuration properties.
    
    Returns:
    
    Since:
    
    0.4.0
  - setKafkaProducerProperties
```
public KafkaQueue setKafkaProducerProperties(Properties kafkaProducerConfigs)
```
    Sets Kafka producer's custom configuration properties.
    
    Parameters:
    
    kafkaProducerConfigs -
    
    Returns:
    
    Since:
    
    0.4.0
  - getKafkaConsumerProperties
```
public Properties getKafkaConsumerProperties()
```
    Gets Kafka consumer's custom configuration properties.
    
    Returns:
    
    Since:
    
    0.4.0
  - setKafkaConsumerProperties
```
public KafkaQueue setKafkaConsumerProperties(Properties kafkaConsumerConfigs)
```
    Sets Kafka consumer's custom configuration properties.
    
    Parameters:
    
    kafkaConsumerConfigs -
    
    Returns:
    
    Since:
    
    0.4.0
  - getTopicName
```
public String getTopicName()
```
    Name of Kafka topic to store queue messages.
    
    Returns:
  - setTopicName
```
public KafkaQueue setTopicName(String topicName)
```
  - getConsumerGroupId
```
public String getConsumerGroupId()
```
    Kafka's group-id to consume messages.
    
    Returns:
  - setConsumerGroupId
```
public KafkaQueue setConsumerGroupId(String consumerGroupId)
```
  - 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, KafkaQueue 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.
  - close
```
public void close()
```
    Specified by:
    
    close in interface Closeable
    
    Specified by:
    
    close in interface AutoCloseable
    
    Since:
    
    0.4.0
  - 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, partitioning message by IQueueMessage.qId()
    
    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

getKafkaBootstrapServers

setKafkaBootstrapServers

getKafkaProducerProperties

setKafkaProducerProperties

getKafkaConsumerProperties

setKafkaConsumerProperties

getTopicName

setTopicName

getConsumerGroupId

setConsumerGroupId

getKafkaClient

setKafkaClient

init

destroy

close

serialize

deserialize

takeFromQueue

putToQueue

queue

requeue

requeueSilent

finish

take

getOrphanMessages

moveFromEphemeralToQueueStorage

queueSize

ephemeralSize