services.moleculer.transporter

Class KafkaTransporter

java.lang.Object

services.moleculer.service.MoleculerComponent

services.moleculer.transporter.Transporter

services.moleculer.transporter.KafkaTransporter

All Implemented Interfaces:

MetricConstants, MoleculerLifecycle

public class KafkaTransporter
extends Transporter

Kafka Transporter. Kafka is used for building real-time data pipelines and streaming apps. It is horizontally scalable, fault-tolerant, wicked fast, and runs in production in thousands of companies. Sample of usage:

 KafkaTransporter kafka = new KafkaTransporter();
 kafka.setUrls(new String[] { "192.168.51.29:9092" });
 kafka.setDebug(true);
 kafka.setProducerProperty("session.timeout.ms", "30000");
 ServiceBroker broker = ServiceBroker.builder().transporter(kafka).build();
 // broker.createService(new Service("test") {...});
 broker.start();
 

Required dependency:

// https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients
compile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.4.0'

See Also:

TcpTransporter, RedisTransporter, NatsTransporter, NatsStreamingTransporter, MqttTransporter, JmsTransporter, AmqpTransporter

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	KafkaTransporter.KafkaPoller

Field Summary

Fields

Modifier and Type	Field and Description
protected java.util.Properties	consumerProperties
protected java.util.concurrent.ExecutorService	executor Executor of reader loop for incoming messages
protected KafkaTransporter.KafkaPoller	poller
protected org.apache.kafka.clients.producer.KafkaProducer<byte[],byte[]>	producer
protected java.util.Properties	producerProperties
protected boolean	shutDownThreadPools
protected java.lang.String[]	urls

Fields inherited from class services.moleculer.transporter.Transporter

checkTimeoutTimer, counterTransporterPacketsReceivedBytes, counterTransporterPacketsReceivedTotal, counterTransporterPacketsSentBytes, counterTransporterPacketsSentTotal, debug, debugHeartbeats, disconnectChannel, discoverBroadcastChannel, discoverChannel, eventbus, eventChannel, gaugeTransitConnected, heartbeatChannel, heartbeatInterval, heartbeatTimeout, heartBeatTimer, infoBroadcastChannel, infoChannel, infoScheduled, infoScheduledAt, instanceID, lastReceivedMessageAt, metrics, monitor, namespace, nodeID, nodes, offlineTimeout, PACKET_DISCONNECT, PACKET_DISCOVER, PACKET_EVENT, PACKET_HEARTBEAT, PACKET_INFO, PACKET_PING, PACKET_PONG, PACKET_REQUEST, PACKET_RESPONSE, pingChannel, pongChannel, preferHostname, prefix, protocolVersion, registry, requestChannel, responseChannel, scheduler, serializer, serviceInvoker, subscriptionTimeout, uidGenerator, usingJsonSerializer

Fields inherited from class services.moleculer.service.MoleculerComponent

broker, logger, name

Fields inherited from interface services.moleculer.metrics.MetricConstants

MOLECULER_CACHER_CLEAN_TIME, MOLECULER_CACHER_CLEAN_TIME_DESC, MOLECULER_CACHER_CLEAN_TOTAL, MOLECULER_CACHER_CLEAN_TOTAL_DESC, MOLECULER_CACHER_DEL_TIME, MOLECULER_CACHER_DEL_TIME_DESC, MOLECULER_CACHER_DEL_TOTAL, MOLECULER_CACHER_DEL_TOTAL_DESC, MOLECULER_CACHER_EXPIRED_TOTAL, MOLECULER_CACHER_EXPIRED_TOTAL_DESC, MOLECULER_CACHER_FOUND_TOTAL, MOLECULER_CACHER_FOUND_TOTAL_DESC, MOLECULER_CACHER_GET_TIME, MOLECULER_CACHER_GET_TIME_DESC, MOLECULER_CACHER_GET_TOTAL, MOLECULER_CACHER_GET_TOTAL_DESC, MOLECULER_CACHER_SET_TIME, MOLECULER_CACHER_SET_TIME_DESC, MOLECULER_CACHER_SET_TOTAL, MOLECULER_CACHER_SET_TOTAL_DESC, MOLECULER_CIRCUIT_BREAKER_HALF_OPENED_ACTIVE, MOLECULER_CIRCUIT_BREAKER_HALF_OPENED_ACTIVE_DESC, MOLECULER_CIRCUIT_BREAKER_OPENED_ACTIVE, MOLECULER_CIRCUIT_BREAKER_OPENED_ACTIVE_DESC, MOLECULER_CIRCUIT_BREAKER_OPENED_TOTAL, MOLECULER_CIRCUIT_BREAKER_OPENED_TOTAL_DESC, MOLECULER_EVENT_BROADCAST_TOTAL, MOLECULER_EVENT_BROADCAST_TOTAL_DESC, MOLECULER_EVENT_BROADCASTLOCAL_TOTAL, MOLECULER_EVENT_BROADCASTLOCAL_TOTAL_DESC, MOLECULER_EVENT_EMIT_TOTAL, MOLECULER_EVENT_EMIT_TOTAL_DESC, MOLECULER_EVENT_RECEIVED_ACTIVE, MOLECULER_EVENT_RECEIVED_ACTIVE_DESC, MOLECULER_EVENT_RECEIVED_ERROR_TOTAL, MOLECULER_EVENT_RECEIVED_ERROR_TOTAL_DESC, MOLECULER_EVENT_RECEIVED_TIME, MOLECULER_EVENT_RECEIVED_TIME_DESC, MOLECULER_EVENT_RECEIVED_TOTAL, MOLECULER_EVENT_RECEIVED_TOTAL_DESC, MOLECULER_EXECUTOR, MOLECULER_REQUEST_ACTIVE, MOLECULER_REQUEST_ACTIVE_DESC, MOLECULER_REQUEST_ERROR_TOTAL, MOLECULER_REQUEST_ERROR_TOTAL_DESC, MOLECULER_REQUEST_LEVELS, MOLECULER_REQUEST_LEVELS_DESC, MOLECULER_REQUEST_RETRY_ATTEMPTS_TOTAL, MOLECULER_REQUEST_RETRY_ATTEMPTS_TOTAL_DESC, MOLECULER_REQUEST_TIME, MOLECULER_REQUEST_TIME_DESC, MOLECULER_REQUEST_TIMEOUT_TOTAL, MOLECULER_REQUEST_TIMEOUT_TOTAL_DESC, MOLECULER_REQUEST_TOTAL, MOLECULER_REQUEST_TOTAL_DESC, MOLECULER_SCHEDULER, MOLECULER_TRANSIT_CONNECTED, MOLECULER_TRANSIT_CONNECTED_DESC, MOLECULER_TRANSIT_ORPHAN_RESPONSE_TOTAL, MOLECULER_TRANSIT_ORPHAN_RESPONSE_TOTAL_DESC, MOLECULER_TRANSIT_REQUESTS_ACTIVE, MOLECULER_TRANSIT_REQUESTS_ACTIVE_DESC, MOLECULER_TRANSIT_STREAMS_RECEIVE_ACTIVE, MOLECULER_TRANSIT_STREAMS_RECEIVE_ACTIVE_DESC, MOLECULER_TRANSPORTER_PACKETS_RECEIVED_BYTES, MOLECULER_TRANSPORTER_PACKETS_RECEIVED_BYTES_DESC, MOLECULER_TRANSPORTER_PACKETS_RECEIVED_TOTAL, MOLECULER_TRANSPORTER_PACKETS_RECEIVED_TOTAL_DESC, MOLECULER_TRANSPORTER_PACKETS_SENT_BYTES, MOLECULER_TRANSPORTER_PACKETS_SENT_BYTES_DESC, MOLECULER_TRANSPORTER_PACKETS_SENT_TOTAL, MOLECULER_TRANSPORTER_PACKETS_SENT_TOTAL_DESC

Constructor Summary

Constructors

Constructor and Description
KafkaTransporter()
KafkaTransporter(java.lang.String... urls)

Method Summary

Modifier and Type	Method and Description
void	connect()
protected void	disconnect()
protected void	error(java.lang.Throwable cause) Any I/O error occurred.
java.util.Properties	getConsumerProperties()
java.util.concurrent.ExecutorService	getExecutor()
java.util.Properties	getProducerProperties()
java.lang.String[]	getUrls()
boolean	isShutDownThreadPools()
void	publish(java.lang.String channel, io.datatree.Tree message)
protected void	reconnect(java.lang.Throwable cause)
void	setConsumerProperties(java.util.Properties consumerProperties)
void	setConsumerProperty(java.lang.String key, java.lang.String value)
void	setExecutor(java.util.concurrent.ExecutorService executor)
void	setProducerProperties(java.util.Properties producerProperties)
void	setProducerProperty(java.lang.String key, java.lang.String value)
void	setShutDownThreadPools(boolean shutDownThreadPools)
void	setUrls(java.lang.String[] urls)
void	stopped() Closes transporter.
io.datatree.Promise	subscribe(java.lang.String channel)

Methods inherited from class services.moleculer.transporter.Transporter

broadcastInfoPacket, broadcastNodeConnected, broadcastNodeDisconnected, broadcastNodeUpdated, broadcastTransporterConnected, broadcastTransporterDisconnected, channel, checkTimeouts, connected, connected, createPingPacket, getAddress, getAllNodeIDs, getCpuUsage, getDescriptor, getHeartbeatInterval, getHeartbeatTimeout, getInstanceID, getLastHeartbeatTime, getNamespace, getOfflineTimeout, getPrefix, getSerializer, getSubscriptionTimeout, isDebug, isDebugHeartbeats, isOnline, isPreferHostname, processReceivedMessage, processReceivedMessage, publish, received, scheduleInfoPacket, sendClosePacket, sendDataPacket, sendDisconnectPacket, sendDiscoverPacket, sendErrorPacket, sendEventPacket, sendHeartbeatPacket, sendInfoPacket, sendPongPacket, sendRequestPacket, setDebug, setDebugHeartbeats, setHeartbeatInterval, setHeartbeatTimeout, setNamespace, setOfflineTimeout, setPreferHostname, setPrefix, setSerializer, setSubscriptionTimeout, started, subscribe, updateNodeInfo

Methods inherited from class services.moleculer.service.MoleculerComponent

getBroker, getLogger, getName

Methods inherited from class java.lang.Object

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

Field Detail

producerProperties

protected java.util.Properties producerProperties

consumerProperties

protected java.util.Properties consumerProperties

urls

protected java.lang.String[] urls

producer

protected org.apache.kafka.clients.producer.KafkaProducer<byte[],byte[]> producer

poller

protected KafkaTransporter.KafkaPoller poller

executor

protected java.util.concurrent.ExecutorService executor

Executor of reader loop for incoming messages

shutDownThreadPools

protected boolean shutDownThreadPools

Constructor Detail

KafkaTransporter

public KafkaTransporter()

KafkaTransporter

public KafkaTransporter(java.lang.String... urls)

Method Detail

connect

public void connect()

Specified by:

connect in class Transporter

disconnect

protected void disconnect()

reconnect

protected void reconnect(java.lang.Throwable cause)

error

protected void error(java.lang.Throwable cause)

Description copied from class: Transporter

Any I/O error occurred. Implementation-specific error handling goes here (reconnection, etc.).

Overrides:

error in class Transporter

Parameters:

cause - I/O error

stopped

public void stopped()

Closes transporter.

Specified by:

stopped in interface MoleculerLifecycle

Overrides:

stopped in class Transporter

subscribe

public io.datatree.Promise subscribe(java.lang.String channel)

Specified by:

subscribe in class Transporter

publish

public void publish(java.lang.String channel,
                    io.datatree.Tree message)

Specified by:

publish in class Transporter

setProducerProperty

public void setProducerProperty(java.lang.String key,
                                java.lang.String value)

setConsumerProperty

public void setConsumerProperty(java.lang.String key,
                                java.lang.String value)

getUrls

public java.lang.String[] getUrls()

setUrls

public void setUrls(java.lang.String[] urls)

getProducerProperties

public java.util.Properties getProducerProperties()

setProducerProperties

public void setProducerProperties(java.util.Properties producerProperties)

getConsumerProperties

public java.util.Properties getConsumerProperties()

setConsumerProperties

public void setConsumerProperties(java.util.Properties consumerProperties)

getExecutor

public java.util.concurrent.ExecutorService getExecutor()

setExecutor

public void setExecutor(java.util.concurrent.ExecutorService executor)

isShutDownThreadPools

public boolean isShutDownThreadPools()

setShutDownThreadPools

public void setShutDownThreadPools(boolean shutDownThreadPools)