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 adminRequestSend(node,request) {

	if(logger.active) logger.send({label:'adminRequestSend',action:request.action,properties:Object.keys(request)})

		if(request.data) node.connection[request.action](request.data,

//					if(logger.active) logger.send({label:'adminRequestSend connection',error:err,data:data})

		logger.send({label:'adminRequestProcess',error:ex.message,stack:ex.stack});

	if(logger.active) logger.send({label:'adminRequest',action:request.action,properties:Object.keys(request)})

	if(!request.action) throw Error("action not provided")

	if(!request.callback) throw Error("callback not provided")

	if(!request.error) throw Error("error function not provided")

		this.processInput(this,request) // processInput=adminRequestSend

		logger.info(this.waiting.length + ' in wait queue')

		logger.error('adminRequest connection failed, clearing waiting queue ' + node.waiting.length)

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

  if(logger.active) logger.send({label: 'hostAvailable',host: host,port: port,node: node.id});

	if(logger.active) logger.send({label:'hostAvailable connect',host:host,port:port,node: node.id})

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

      if(logger.active) logger.send({label: 'hostAvailable end',host: host,port: port,node: node.id});

      if(logger.active) logger.send({label: 'hostAvailable error',host: host,port: port,node: node.id,error: e.toString()});

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

      if(logger.active) logger.send({label:'hostAvailable timeout',host:host,port:port,node:node.id});

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

  if(logger.active) logger.send({label:'testHosts',node:node.id,available: node.available});

    	node.runtimeTimer = setInterval(function () {

          node.send('runtimeTimer ' + e.message)

            testHosts.apply(node, [node])

            node.send('runtimeTimer ' + e.message)

      this.close = function (removed, done) {

      node.close = function (removed, done) {

   	    node.onCloseStack.forEach(callFunction=>callFunction())

        node.metadata=new Metadata(node,logger);

        node.onConnectStack.push(node.metadata.startRefresh);

        node.onCloseStack.push(node.metadata.stopRefresh);

        node.onChangeMetadata=node.metadata.onChange.bind(node.metadata);

        node.metadataRefresh=node.metadata.refresh.bind(node.metadata);

        node.getTopicsPartitions=node.metadata.getTopicsPartitions.bind(node.metadata);

    	logger.error({label:"metadata",error:ex.message,stack:ex.stack})

  KafkaBrokerNode.prototype.close = function (removed, done) {

  RED.nodes.registerType(logger.label, KafkaBrokerNode, {

  "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.

* Generic topic(s) for a consumer using regex which dynamically adds new topics as they are defined

* For consumer add/remove topics (not persisted)

* Convert "/" to "." to assist with interfaces to other queueing technologies

Note: all nodes run in debug mode for 111 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. 

If wildcard selected then topics are regex patterns which are dynamically made active (or removed) when available.

A check is performed once every minute for chanaes to topics.

![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")

If msg._kafka exists and the consumer associated with the message is not on autocommit, it closes the consumer.  This effectively rolls back the message in Kafka plus ensures the message cannot be automatically handed to the the consumer.  It is expected that the message or problem is fixed and the consumer opened again for processing.

Run the following command in the root directory of your Node-RED install or via GUI install

    npm install node-red-contrib-kafka-manager

		@@ -366,15 +366,15 @@ const logger = new (require("node-red-contrib-logger"))("Kafka Broker");
		node.runtimeTimer = setInterval(function () {
		try {
		testHosts.apply(node, [node])
		} catch (e) {
		node.send('runtimeTimer ' + e.message)
		}
		},
		node.checkInterval * 1000
		try {
		testHosts.apply(node, [node])
		} catch (e) {
		node.send('runtimeTimer ' + e.message)
		}
		},
		node.checkInterval * 1000
		)
		this.close = function (removed, done) {
		logger.info("close")
		node.close = function (removed, done) {
		logger.info("close")
		node.onCloseStack.forEach(callFunction=>callFunction())
		runtimeStop.apply(node)
		done()
		if(done) done()
		}
		@@ -381,0 +381,0 @@ }

		{
		"name": "node-red-contrib-kafka-manager",
		"version": "0.5.0",
		"version": "0.5.1",
		"description": "Node-RED implements Kafka manager with associand associated .",
		@@ -12,3 +12,3 @@ "dependencies": {
		"bl": ">=2.2.1",
		"eslint": "^6.8.0",
		"eslint": "^8.9.0",
		"eslint-config-standard": "^14.1.0",
		@@ -19,4 +19,4 @@ "eslint-plugin-import": "^2.22.1",
		"eslint-plugin-standard": "^4.0.1",
		"mocha": "^8.2.0",
		"node-red": "^1.2.2",
		"mocha": "^9.2.0",
		"node-red": "^2.2.1",
		"node-red-node-test-helper": "*"
		@@ -23,0 +23,0 @@ },

		@@ -31,3 +31,3 @@ # [node-red-contrib-kafka-manager][2]

		in settitngs.js
		in settings.js

		@@ -34,0 +34,0 @@ ![Kafka Broker](documentation/broker.JPG "Kafka Broker")

Improved metrics