Node-RED implements Kafka manager with associand associated . 

peter.prib

node-red-contrib-kafka-manager

const logger = new (require("node-red-contrib-logger"))("Kafka Broker");

function hostAvailable (host, port, node, availableCB, downCB, timeoutCB) {

      node.error('hostAvailable on connect error ' + e.message)

        node.error('hostAvailable on error ' + e.message)

        (downCB || timeoutCB).apply(node, ['time out'])

  hostAvailable(host.host, host.port, node,

        if(node.hostsEnvVar in process.env) {

            	const hosts=JSON.parse(process.env[node.hostsEnvVar]);

            		logger.send({label:"test",hosts:hosts});

            		if(hosts instanceof Array) {

            			node.hostsCombined=node.hostsCombined.concat(hosts);

            			throw Error("not array value: "+process.env[node.hostsEnvVar]);

    			const error="process.env variable "+node.hostsEnvVar+e.toString();

        	const error="process.env."+node.hostsEnvVar+" not found ";

    if (node.hosts.length === 0 && node.host) {

    node.hostsCombined=node.hostsCombined.concat(node.hosts);

    logger.send({hosts:node.hostsCombined});

    node.kafkaHost = node.hostsCombined.map((r) => r.host + ':' + r.port).join(',')

        connectTimeout: node.connectTimeout || 10000,

        requestTimeout: node.requestTimeout || 30000,

        autoConnect: (node.autoConnect || 'true') === 'true',

        idleConnection: node.idleConnection || 5,

        reconnectOnIdle: (node.reconnectOnIdle || 'true') === 'true',

        maxAsyncRequests: node.maxAsyncRequests || 10,

        useCredentials: node.useCredentials || false

   	  if(logger.active) logger.send({label:'getKafkaClient',usetls:node.usetls,options:options});

    	  options.sslOptions = {rejectUnauthorized:node.selfSign};

      	  if(logger.active) logger.send({label:'getKafkaClient tls use',selfSign:node.selfSign});

    		  if(!(node.selfServe||node.tls)) throw Error("not self serve or no tls configuration selected");

        		  node.tlsNode = RED.nodes.getNode(node.tls);

     	  	      if (!node.tlsNode) throw Error("tls configuration not found");

      	  	      Object.assign(options.sslOptions,node.tlsNode.credentials);

      	  	      if(logger.active) logger.send({label:'getKafkaClient sslOptions',properties:Object.keys(options.sslOptions)});

    		  node.error("get node tls "+node.tls+" failed, error:"+e);

    	if(logger.active) logger.send({label:'getKafkaClient node has configured credentials, note sasl mechanism is plain'});

          username: this.credentials.user,

          password: node.credentials.password

        options: Object.assign({},options,options.sslOptions?{sslOptions:"***masked***"}:null)

      return new kafka.KafkaClient(options)

      if (typeof err === 'string' && err.startsWith('connect ECONNREFUSED')) return 'Connection refused, check if Kafka up'

  "name": "node-red-contrib-kafka-manager",

  "description": "Node-RED implements Kafka manager with associand associated .",

    "lint": "eslint --ext .js kafkaManager",

    "lint-fix": "eslint --fix --ext .js kafkaManager"

    "url": "git+https://github.com/peterprib/node-red-contrib-kafka-manager.git"

      "kafkaAdmin": "kafkaManager/kafkaAdmin.js",

      "kafkaBroker": "kafkaManager/kafkaBroker.js",

      "kafkaCommit": "kafkaManager/kafkaCommit.js",

      "kafkaConsumer": "kafkaManager/kafkaConsumer.js",

      "kafkaConsumerGroup": "kafkaManager/kafkaConsumerGroup.js",

      "kafkaOffset": "kafkaManager/kafkaOffset.js",

      "kafkaProducer": "kafkaManager/kafkaProducer.js",

      "kafkaRollback": "kafkaManager/kafkaRollback.js"

    "url": "https://github.com/peterprib/node-red-contrib-kafka-manager/issues"

  "homepage": "https://github.com/peterprib/node-red-contrib-kafka-manager#readme"

[Node Red][1] for working with apache kafka, a streaming product.

First initial release using [kafka-node][4] .

Has a test GUI which allows topics to be added.

Note: all nodes run in debug mode for 100 messages then turns off.

------------------------------------------------------------

Defines the client interface to kafka. One can add process.env for hosts with 

	process.env.atesthosts='[{"host":"atesthost1","port":1234},{"host":"atesthost2","port":4321}]';

![Kafka Broker](documentation/broker.JPG "Kafka Broker")

![Kafka Broker Options](documentation/brokerOptions.JPG "Kafka Broker Options")

Provide the ability to process administration tasks such as create and list topic. 

Following topics commnads or via GUI allowed:

Plus following topics commands allowed where paramaters are payload:

*   alterConfigs, alterReplicaLogDirs, createAcls,createDelegationToke

*   createPartitions, createTopics, deleteAcls, deleteConsumerGroups, deleteRecords

*   deleteTopics, describeAcls, describeConsumerGroups, describeGroups, describeLogDirs, describeTopics   *   electPreferredLeaders, expireDelegationToken, incrementalAlterConfigs, listConsumerGroupOffsets, renewDelegationToken

![Admin](documentation/admin.JPG "Admin")

If msg._kafka exists and the consumer associated with the message is not on autocommit, it issues commit for the consumer that had produced the message.  Sends to OK or Error port depending on state.

Note, as Kakfa keeps giving messages to consumer regardless if commit being outstanding, the commit may commit many in-flight messages.  Haven't identified a method of readily preventing this behavour without complications.

Consumer of topic messages in kafka which are generated into node-red message. 

![Kafka Consumer](documentation/consumer.JPG "Kafka Consumer")

![Kafka Consumer Options](documentation/consumerOptions.JPG "Kafka Consumer Options")

![Kafka Consumer Fetch](documentation/consumerFetch.JPG "Kafka Consumer Fetch")

![Kafka Consumer Encoding](documentation/consumerEncoding.JPG "Kafka Consumer Encoding")

![Kafka Consumer Group](documentation/consumerGroup.JPG "Kafka Consumer Group")

![Kafka Consumer Group Options](documentation/consumerGroupOptions.JPG "Kafka Consumer Options Group")

Get various offsets from Kafka. Which type are set via msg.action or msg.topic.  msg.payload states the types of options.

![Kafka Offset](documentation/offset.JPG "Kafka Offset")

Converts a node-red message into a kafka messages.

![Kafka Producer](documentation/producer.JPG "Kafka Producer")

0.4.0 Add commit and rollback with ability to close and start consumer.

0.2.14 Add self serve TLS and fix bug plus mask ssl info when debug logging

0.2.9 Change debugging mechanism and add kafka-node to dependencies

0.2.8 Added all admin api's per Kafka 2.3 but dependent on [kafka-node][4] update.

Remove refresh metadata, automated if problem.  Fix consumer group errors.  Add tests for admin calls.

0.2.7 If offsetOutOfRange pause consumer.  Added in deleteTopics but dependant on [kafka-node][4] update.

0.2.6 More fixes for error processing on invalid topic

0.2.2 Stopped bug where producer on connection initiates a null message. Fix bug with restart logic on fail and order of messages on failed retry

0.2.1 When messages being queued as producer is waiting on connection or reconnection show producer in problem state.

0.2.0 Add nodes consumer group and offset.

	Another fix to issue with initial no kafka.

0.1.0 Add in High level producer/consumer.

  	Further fixes to make connection more robust on kafka up/down

  	Make connection more robust on kafka up/down due to bugs and problem points in in [kafka-node][4] 

[1]: http://nodered.org "node-red home page"

[2]: https://www.npmjs.com/package/node-red-contrib-kafka-manager "source code"

[3]: https://github.com/peterprib "base github"

[4]: https://github.com/SOHU-Co/kafka-node "npm kafka-node"

		@@ -324,2 +324,3 @@ const logger = new (require("node-red-contrib-logger"))("Kafka Broker");
		setState: setState,
		closeNode: closeNode,
		stateUp: [],
		@@ -326,0 +327,0 @@ stateDown: [],

		{
		"name": "node-red-contrib-kafka-manager",
		"version": "0.4.1",
		"version": "0.4.2",
		"description": "Node-RED implements Kafka manager with associand associated .",
		@@ -10,10 +10,12 @@ "dependencies": {
		"devDependencies": {
		"bcrypt": ">=5.0.0",
		"bl": ">=2.2.1",
		"eslint": "^6.8.0",
		"eslint-config-standard": "^14.1.0",
		"eslint-plugin-import": "^2.19.1",
		"eslint-plugin-import": "^2.22.1",
		"eslint-plugin-node": "^11.0.0",
		"eslint-plugin-promise": "^4.2.1",
		"eslint-plugin-standard": "^4.0.1",
		"mocha": "^8.1.1",
		"node-red": "^1.1.3",
		"mocha": "^8.2.0",
		"node-red": "^1.2.2",
		"node-red-node-test-helper": "*"
		@@ -20,0 +22,0 @@ },

		@@ -136,2 +136,4 @@ # [node-red-contrib-kafka-manager][2]

		0.4.1 minor fixes

		0.4.0 Add commit and rollback with ability to close and start consumer.
		@@ -169,5 +171,2 @@

		0.0.2 Add in High level producer.
		Make connection more robust on kafka up/down due to bugs and problem points in in [kafka-node][4]

		0.0.1 base
		@@ -174,0 +173,0 @@

Improved metrics

Worsened metrics