Implements: MessageQueue, IReferenceable, IUnreferenceable, IConfigurable, IOpenable, ICleanable
Description
The KafkaMessageQueue class allows you to create message queues that send and receive messages via the Kafka message broker.
Configuration parameters
- topic: name of Kafka topic to subscribe
- group_id: (optional) consumer group id (default: default)
- from_beginning: (optional) restarts receiving messages from the beginning (default: false)
- read_partitions: (optional) number of partitions to be consumed concurrently (default: 1)
- autocommit: (optional) turns on/off autocommit (default: true)
- connection(s):
- discovery_key: (optional) key to retrieve the connection from IDiscovery
- host: host name or IP address
- port: port number
- uri: resource URI or connection string with all parameters in it
- credential(s):
- store_key: (optional) key to retrieve the credentials from ICredentialStore
- username: username
- password: user’s password
- options:
- autosubscribe: (optional) true to automatically subscribe on option (default: false)
- acks: (optional) control the number of required acks: -1 - all, 0 - none, 1 - only leader (default: -1)
- log_level: (optional) log level 0 - None, 1 - Error, 2 - Warn, 3 - Info, 4 - Debug (default: 1)
- connect_timeout: (optional) number of milliseconds to connect to broker (default: 1000)
- max_retries: (optional) maximum retry attempts (default: 5)
- retry_timeout: (optional) number of milliseconds to wait on each reconnection attempt (default: 30000)
- request_timeout: (optional) number of milliseconds to wait on flushing messages (default: 30000)
References
- *:logger:*:*:1.0 - (optional) ILogger components to pass log messages
- *:counters:*:*:1.0 - (optional) ICounters components to pass collected measurements
- *:discovery:*:*:1.0 - (optional) IDiscovery services
- *:credential-store:*:*:1.0 - (optional) ICredentialStore to resolve credentials
- *:connection:kafka:*:1.0 - (optional) shared connection to Kafka service
Constructors
Creates a new instance of the message queue.
KafkaMessageQueue(name: str = None)
- name: str - (optional) queue name.
Fields
Instance methods
abandon
Returns a message into the queue and makes it available for all subscribers to receive it again. This method is usually used to return a message which could not be processed at the moment to repeat the attempt. Messages that cause unrecoverable errors shall be removed permanently or/and sent to dead letter queue.
abandon(message: MessageEnvelope)
- message: MessageEnvelope - message to return.
clear
Clears a component’s state.
clear(correlation_id: Optional[str])
- correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
close
Closes a component and frees used resources.
close(correlation_id: Optional[str])
- correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
complete
Permanently removes a message from the queue. This method is usually used to remove the message after successful processing.
complete(message: MessageEnvelope)
- message: MessageEnvelope - message to remove.
configure
Configures a component by passing its configuration parameters.
configure(config: ConfigParams)
- config:: ConfigParams - configuration parameters to be set.
endListen
Ends listening for incoming messages. When this method is called, listen unblocks the thread and execution continues.
endListen(correlation_id: Optional[str])
- correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
_from_message
Returns the content of a message and information about it.
_from_message(message: MessageEnvelope): Optional[dict]
- message: MessageEnvelope - message
- returns: Optional[dict] - content of the message and information about it.
_get_topic
Returns the topic.
_get_topic(): str
- returns: str - topic
is_open
Checks if the component is open.
is_open(): bool
- returns: bool - true if the component is open and false otherwise.
listen
Listens for incoming messages and blocks the current thread until the queue is closed.
See IMessageReceiver
listen(correlation_id: Optional[str], receiver: IMessageReceiver)
- correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
- receiver: IMessageReceiver - receiver used to receive incoming messages.
move_to_dead_letter
Permanently removes a message from the queue and sends it to dead letter queue.
- Important: This method is not supported by Kafka.
move_to_dead_letter(message: MessageEnvelope)
- message: MessageEnvelope - message to be removed.
on_message
Deserializes a message. Then, sends it to a receiver if its set or puts it into the queue.
on_message(topic: str, partition: int, msg: Message)
- topic: str - topic
- partition: int - partition number
- msg: Message - message
open
Opens the component.
open(correlation_id: Optional[str])
- correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
peek
Peeks a single incoming message from the queue without removing it. If there are no messages available in the queue, it returns null.
peek(correlation_id: Optional[str]): MessageEnvelope
- correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
- returns: MessageEnvelope - peeked message.
peek_batch
Peeks multiple incoming messages from the queue without removing them. If there are no messages available in the queue, it returns an empty list.
peek_batch(correlation_id: Optional[str], messageCount: number): List[MessageEnvelope]
- correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
- messageCount: number - maximum number of messages to peek.
- returns: List[MessageEnvelope] - list with peeked messages.
read_message_count
Reads the current number of messages in the queue to be delivered.
read_message_count(): int
- *returns: int - number of messages in the queue.
receive
Receives an incoming message and removes it from the queue.
receive(correlation_id: Optional[str], waitTimeout: number): MessageEnvelope
- correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
- waitTimeout: number - timeout in milliseconds to wait for a message to come.
- returns: MessageEnvelope - received message or null if nothing was received.
renew_lock
Renews a lock on a message that makes it invisible from other receivers in the queue. This method is usually used to extend the message processing time.
- Important: This method is not supported by Kafka.
renew_lock(message: MessageEnvelope, lock_timeout: int)
- message: MessageEnvelope - message to extend its lock.
- lock_timeout: int - locking timeout in milliseconds.
send
Sends a message into the queue.
send(correlation_id: Optional[str], message: MessageEnvelope)
- correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
- message: MessageEnvelope - message envelop to be sent.
set_references
Sets references to dependent components.
set_references(references: IReferences)
- references: IReferences - references to locate the component dependencies.
_subscribe
Subscribes to a topic.
_subscribe(correlation_id: Optional[str])
- correlation_id: (optional) transaction id used to trace execution through the call chain.
_to_message
Creates a new MessageEnvelope.
_to_message(msg: Message): Optional[MessageEnvelope]
- msg: Message - message
- returns: MessageEnvelope - message
unset_references
Unsets (clears) previously set references to dependent components.
unset_references()
Examples
queue = KafkaMessageQueue("myqueue")
queue.configure(ConfigParams.from_tuples(
"topic", "mytopic",
'connection.protocol', 'tcp',
"connection.host", "localhost",
"connection.port", 9092,
))
queue.open("123")
queue.send("123", MessageEnvelope(None, "mymessage", "ABC"))
message = queue.receive("123", 10000)
print(message)
if message is not None:
...
queue.complete(message)