kafkajs - npm Package Compare versions

Comparing version 1.4.4 to 1.4.5

CHANGELOG.md

@@ -8,2 +8,10 @@ # Changelog
 
+## [1.4.5] - 2018-11-28
+
+### Fixed
+  - Fix crash in mitigation for receiving metadata for unsubscribed topics #221
+
+### Added
+  - Add `CRASH` instrumentation event for the consumer #221
+
 ## [1.4.4] - 2018-10-29
@@ -10,0 +18,0 @@

package.json

 {
   "name": "kafkajs",
-  "version": "1.4.4",
+  "version": "1.4.5",
   "description": "A modern Apache Kafka client for node.js",
@@ -5,0 +5,0 @@ "author": "Tulio Ornelas <[e-mail]>",

README.md

@@ -1158,2 +1158,5 @@ [![KafkaJS](https://raw.githubusercontent.com/tulios/kafkajs/master/logo.png)](https://github.com/tulios/kafkajs)
 
+* consumer.events.CRASH
+  payload: {`error`, `groupId`}
+
 ### <a name="instrumentation-producer"></a> Producer
@@ -1160,0 +1163,0 @@

src/consumer/consumerGroup.js

@@ -169,3 +169,4 @@ const flatten = require('../utils/flatten')
     // Check if the consumer is aware of all assigned partitions
-    for (let topic of assignedTopics) {
+    const safeAssignedTopics = keys(currentMemberAssignment)
+    for (let topic of safeAssignedTopics) {
       const assignedPartitions = currentMemberAssignment[topic]
@@ -172,0 +173,0 @@ const knownPartitions = this.partitionsPerSubscribedTopic.get(topic)

src/consumer/index.js

@@ -7,3 +7,3 @@ const Long = require('long')
 const InstrumentationEventEmitter = require('../instrumentation/emitter')
-const { CONNECT, DISCONNECT, STOP } = require('./instrumentationEvents')
+const { CONNECT, DISCONNECT, STOP, CRASH } = require('./instrumentationEvents')
 const { KafkaJSNonRetriableError } = require('../errors')
@@ -190,2 +190,7 @@ const { roundRobin } = require('./assigners')
 
+      instrumentationEmitter.emit(CRASH, {
+        error: e,
+        groupId,
+      })
+
       if (e.name === 'KafkaJSNumberOfRetriesExceeded') {
@@ -192,0 +197,0 @@ logger.error(`Restarting the consumer in ${e.retryTime}ms`, {

src/consumer/instrumentationEvents.js

@@ -14,2 +14,3 @@ const InstrumentationEventType = require('../instrumentation/eventType')
   STOP: consumerType('stop'),
+  CRASH: consumerType('crash'),
 }

No alert changes

Improved metrics

Total package byte prevSize

485301

increased by0.09%

Lines of code

9100

increased by0.07%

Number of lines in readme file

1334

increased by0.23%

No dependency changes