kafkajs - npm Package Compare versions

Comparing version 1.15.0-beta.14 to 1.15.0-beta.15

package.json

 {
   "name": "kafkajs",
-  "version": "1.15.0-beta.14",
+  "version": "1.15.0-beta.15",
   "description": "A modern Apache Kafka client for node.js",
@@ -83,5 +83,5 @@ "author": "Tulio Ornelas <[e-mail]>",
   "kafkajs": {
-    "sha": "f7a166488321216a0feec4428f8e589b116eb31f",
-    "compare": "https://github.com/tulios/kafkajs/compare/v1.14.0...f7a166488321216a0feec4428f8e589b116eb31f"
+    "sha": "0868189d7388a2868bbc241b1bc763f4f570d032",
+    "compare": "https://github.com/tulios/kafkajs/compare/v1.14.0...0868189d7388a2868bbc241b1bc763f4f570d032"
   }
 }

src/protocol/requests/addOffsetsToTxn/index.js

@@ -7,2 +7,7 @@ const versions = {
   },
+  1: ({ transactionalId, producerId, producerEpoch, groupId }) => {
+    const request = require('./v1/request')
+    const response = require('./v1/response')
+    return { request: request({ transactionalId, producerId, producerEpoch, groupId }), response }
+  },
 }
@@ -9,0 +14,0 @@

src/protocol/requests/addPartitionsToTxn/index.js

@@ -7,2 +7,7 @@ const versions = {
   },
+  1: ({ transactionalId, producerId, producerEpoch, topics }) => {
+    const request = require('./v1/request')
+    const response = require('./v1/response')
+    return { request: request({ transactionalId, producerId, producerEpoch, topics }), response }
+  },
 }
@@ -9,0 +14,0 @@

src/protocol/requests/alterConfigs/index.js

@@ -7,2 +7,7 @@ const versions = {
   },
+  1: ({ resources, validateOnly }) => {
+    const request = require('./v1/request')
+    const response = require('./v1/response')
+    return { request: request({ resources, validateOnly }), response }
+  },
 }
@@ -9,0 +14,0 @@

src/protocol/requests/apiVersions/v2/response.js

@@ -1,4 +0,7 @@
-const { parse: parseV1, decode: decodeV1 } = require('../v1/response')
+const { parse, decode: decodeV1 } = require('../v1/response')
 
 /**
+ * Starting in version 2, on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
+ *
  * ApiVersions Response (Version: 2) => error_code [api_versions] throttle_time_ms
@@ -13,5 +16,15 @@ *   error_code => INT16
 
+const decode = async rawData => {
+  const decoded = await decodeV1(rawData)
+
+  return {
+    ...decoded,
+    throttleTime: 0,
+    clientSideThrottleTime: decoded.throttleTime,
+  }
+}
+
 module.exports = {
-  parse: parseV1,
-  decode: decodeV1,
+  decode,
+  parse,
 }

src/protocol/requests/createAcls/v1/response.js

@@ -1,5 +0,7 @@
-const Decoder = require('../../../decoder')
-const { parse: parseV0 } = require('../v0/response')
+const { parse, decode: decodeV0 } = require('../v0/response')
 
 /**
+ * Starting in version 1, on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
+ *
  * CreateAcls Response (Version: 1) => throttle_time_ms [creation_responses]
@@ -12,15 +14,9 @@ *   throttle_time_ms => INT32
 
-const decodeCreationResponse = decoder => ({
-  errorCode: decoder.readInt16(),
-  errorMessage: decoder.readString(),
-})
-
 const decode = async rawData => {
-  const decoder = new Decoder(rawData)
-  const throttleTime = decoder.readInt32()
-  const creationResponses = decoder.readArray(decodeCreationResponse)
+  const decoded = await decodeV0(rawData)
 
   return {
-    clientSideThrottleTime: throttleTime,
-    creationResponses,
+    ...decoded,
+    throttleTime: 0,
+    clientSideThrottleTime: decoded.throttleTime,
   }
@@ -31,3 +27,3 @@ }
   decode,
-  parse: parseV0,
+  parse,
 }

src/protocol/requests/createPartitions/v1/response.js

@@ -1,1 +0,28 @@
-module.exports = require('../v0/response')
+const { parse, decode: decodeV0 } = require('../v0/response')
+
+/**
+ * Starting in version 1, on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
+ *
+ * CreatePartitions Response (Version: 0) => throttle_time_ms [topic_errors]
+ *   throttle_time_ms => INT32
+ *   topic_errors => topic error_code error_message
+ *     topic => STRING
+ *     error_code => INT16
+ *     error_message => NULLABLE_STRING
+ */
+
+const decode = async rawData => {
+  const decoded = await decodeV0(rawData)
+
+  return {
+    ...decoded,
+    throttleTime: 0,
+    clientSideThrottleTime: decoded.throttleTime,
+  }
+}
+
+module.exports = {
+  decode,
+  parse,
+}

src/protocol/requests/createTopics/index.js

@@ -17,2 +17,7 @@ const versions = {
   },
+  3: ({ topics, validateOnly, timeout }) => {
+    const request = require('./v3/request')
+    const response = require('./v3/response')
+    return { request: request({ topics, validateOnly, timeout }), response }
+  },
 }
@@ -19,0 +24,0 @@

src/protocol/requests/deleteAcls/v1/response.js

@@ -5,2 +5,7 @@ const Decoder = require('../../../decoder')
 /**
+ * Starting in version 1, on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
+ * Version 1 also introduces a new resource pattern type field.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-290%3A+Support+for+Prefixed+ACLs
+ *
  * DeleteAcls Response (Version: 1) => throttle_time_ms [filter_responses]
@@ -47,2 +52,3 @@ *   throttle_time_ms => INT32
   return {
+    throttleTime: 0,
     clientSideThrottleTime: throttleTime,
@@ -49,0 +55,0 @@ filterResponses,

src/protocol/requests/deleteGroups/v1/response.js

@@ -1,3 +0,27 @@
-const responseV0 = require('../v0/response')
+const { parse, decode: decodeV0 } = require('../v0/response')
 
-module.exports = responseV0
+/**
+ * Starting in version 1, on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
+ *
+ * DeleteGroups Response (Version: 1) => throttle_time_ms [results]
+ *  throttle_time_ms => INT32
+ *  results => group_id error_code
+ *    group_id => STRING
+ *    error_code => INT16
+ */
+
+const decode = async rawData => {
+  const decoded = await decodeV0(rawData)
+
+  return {
+    ...decoded,
+    throttleTime: 0,
+    clientSideThrottleTime: decoded.throttleTime,
+  }
+}
+
+module.exports = {
+  decode,
+  parse,
+}

src/protocol/requests/deleteRecords/v1/response.js

 const responseV0 = require('../v0/response')
 
 /**
+ * Starting in version 1, on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
+ *
  * DeleteRecords Response (Version: 1) => throttle_time_ms [topics]
@@ -12,5 +15,2 @@ *  throttle_time_ms => INT32
  *      error_code => INT16
- *
- * note: In version 1 on quota violation, brokers send out responses before throttling.
- * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
  */
@@ -26,2 +26,3 @@
       ...decoded,
+      throttleTime: 0,
       clientSideThrottleTime: decoded.throttleTime,
@@ -28,0 +29,0 @@ }

src/protocol/requests/deleteTopics/v1/response.js

@@ -5,2 +5,5 @@ const Decoder = require('../../../decoder')
 /**
+ * Starting in version 1, on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
+ *
  * DeleteTopics Response (Version: 1) => throttle_time_ms [topic_error_codes]
@@ -22,4 +25,7 @@ *   throttle_time_ms => INT32
   const decoder = new Decoder(rawData)
+  const throttleTime = decoder.readInt32()
+
   return {
-    throttleTime: decoder.readInt32(),
+    throttleTime: 0,
+    clientSideThrottleTime: throttleTime,
     topicErrors: decoder.readArray(topicErrors).sort(topicNameComparator),
@@ -26,0 +32,0 @@ }

src/protocol/requests/describeAcls/v1/response.js

@@ -1,5 +0,10 @@
-const { parse: parseV0 } = require('../v0/response')
+const { parse } = require('../v0/response')
 const Decoder = require('../../../decoder')
 
 /**
+ * Starting in version 1, on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
+ * Version 1 also introduces a new resource pattern type field.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-290%3A+Support+for+Prefixed+ACLs
+ *
  * DescribeAcls Response (Version: 1) => throttle_time_ms error_code error_message [resources]
@@ -41,2 +46,3 @@ *   throttle_time_ms => INT32
   return {
+    throttleTime: 0,
     clientSideThrottleTime: throttleTime,
@@ -51,3 +57,3 @@ errorCode,
   decode,
-  parse: parseV0,
+  parse,
 }

src/protocol/requests/describeConfigs/index.js

@@ -12,2 +12,7 @@ const versions = {
   },
+  2: ({ resources, includeSynonyms }) => {
+    const request = require('./v2/request')
+    const response = require('./v2/response')
+    return { request: request({ resources, includeSynonyms }), response }
+  },
 }
@@ -14,0 +19,0 @@

src/protocol/requests/describeGroups/index.js

@@ -12,2 +12,7 @@ const versions = {
   },
+  2: ({ groupIds }) => {
+    const request = require('./v2/request')
+    const response = require('./v2/response')
+    return { request: request({ groupIds }), response }
+  },
 }
@@ -14,0 +19,0 @@

src/protocol/requests/endTxn/index.js

@@ -10,2 +10,10 @@ const versions = {
   },
+  1: ({ transactionalId, producerId, producerEpoch, transactionResult }) => {
+    const request = require('./v1/request')
+    const response = require('./v1/response')
+    return {
+      request: request({ transactionalId, producerId, producerEpoch, transactionResult }),
+      response,
+    }
+  },
 }
@@ -12,0 +20,0 @@

src/protocol/requests/findCoordinator/index.js

@@ -14,2 +14,7 @@ const COORDINATOR_TYPES = require('../../coordinatorTypes')
   },
+  2: ({ groupId, coordinatorType = COORDINATOR_TYPES.GROUP }) => {
+    const request = require('./v2/request')
+    const response = require('./v2/response')
+    return { request: request({ coordinatorKey: groupId, coordinatorType }), response }
+  },
 }
@@ -16,0 +21,0 @@

src/protocol/requests/heartbeat/v2/response.js

@@ -16,2 +16,3 @@ const { parse, decode: decodeV1 } = require('../v1/response')
     ...decoded,
+    throttleTime: 0,
     clientSideThrottleTime: decoded.throttleTime,
@@ -18,0 +19,0 @@ }

src/protocol/requests/initProducerId/index.js

@@ -7,2 +7,7 @@ const versions = {
   },
+  1: ({ transactionalId, transactionTimeout = 5000 }) => {
+    const request = require('./v1/request')
+    const response = require('./v1/response')
+    return { request: request({ transactionalId, transactionTimeout }), response }
+  },
 }
@@ -9,0 +14,0 @@

src/protocol/requests/joinGroup/v3/response.js

@@ -1,3 +0,2 @@
-const { parse: parseV0 } = require('../v0/response')
-const { decode: decodeV2 } = require('../v2/response')
+const { parse, decode: decodeV2 } = require('../v2/response')
 
@@ -25,2 +24,3 @@ /**
     ...decoded,
+    throttleTime: 0,
     clientSideThrottleTime: decoded.throttleTime,
@@ -32,3 +32,3 @@ }
   decode,
-  parse: parseV0,
+  parse,
 }

src/protocol/requests/joinGroup/v5/response.js

@@ -49,3 +49,3 @@ const Decoder = require('../../../decoder')
   return {
-    throttleTime,
+    throttleTime: 0,
     clientSideThrottleTime: throttleTime,
@@ -52,0 +52,0 @@ errorCode,

src/protocol/requests/leaveGroup/v2/response.js

@@ -16,2 +16,3 @@ const { parse, decode: decodeV1 } = require('../v1/response')
     ...decoded,
+    throttleTime: 0,
     clientSideThrottleTime: decoded.throttleTime,
@@ -18,0 +19,0 @@ }

src/protocol/requests/leaveGroup/v3/response.js

@@ -23,3 +23,3 @@ const Decoder = require('../../../decoder')
 
-  return { throttleTime, clientSideThrottleTime: throttleTime, errorCode, members }
+  return { throttleTime: 0, clientSideThrottleTime: throttleTime, errorCode, members }
 }
@@ -26,0 +26,0 @@

src/protocol/requests/listGroups/v2/response.js

@@ -1,4 +0,7 @@
-const responseV1 = require('../v1/response')
+const { parse, decode: decodeV1 } = require('../v1/response')
 
 /**
+ * In version 2 on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
+ *
  * ListGroups Response (Version: 2) => error_code [groups]
@@ -11,3 +14,15 @@ *   throttle_time_ms => INT32
  */
+const decode = async rawData => {
+  const decoded = await decodeV1(rawData)
 
-module.exports = responseV1
+  return {
+    ...decoded,
+    throttleTime: 0,
+    clientSideThrottleTime: decoded.throttleTime,
+  }
+}
+
+module.exports = {
+  decode,
+  parse,
+}

src/protocol/requests/listOffsets/index.js

@@ -22,2 +22,7 @@ const ISOLATION_LEVEL = require('../../isolationLevel')
   },
+  3: ({ replicaId = REPLICA_ID, isolationLevel = ISOLATION_LEVEL.READ_COMMITTED, topics }) => {
+    const request = require('./v3/request')
+    const response = require('./v3/response')
+    return { request: request({ replicaId, isolationLevel, topics }), response }
+  },
 }
@@ -24,0 +29,0 @@

src/protocol/requests/metadata/index.js

@@ -32,2 +32,7 @@ const versions = {
   },
+  6: ({ topics, allowAutoTopicCreation }) => {
+    const request = require('./v6/request')
+    const response = require('./v6/response')
+    return { request: request({ topics, allowAutoTopicCreation }), response }
+  },
 }
@@ -34,0 +39,0 @@

src/protocol/requests/offsetCommit/v4/response.js

@@ -1,8 +0,6 @@
-const Decoder = require('../../../decoder')
-const { parse } = require('../v3/response')
+const { parse, decode: decodeV3 } = require('../v3/response')
 
 /**
- * Starting in version 4, on quota violation, brokers send out responses
- * before throttling.
- * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-345%3A+Introduce+static+membership+protocol+to+reduce+consumer+rebalances
+ * Starting in version 4, on quota violation, brokers send out responses before throttling.
+ * @see https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
  *
@@ -17,22 +15,13 @@ * OffsetCommit Response (Version: 4) => throttle_time_ms [responses]
  */
+
 const decode = async rawData => {
-  const decoder = new Decoder(rawData)
-  const throttleTime = decoder.readInt32()
+  const decoded = await decodeV3(rawData)
+
   return {
-    throttleTime,
-    clientSideThrottleTime: throttleTime,
-    responses: decoder.readArray(decodeResponses),
+    ...decoded,
+    throttleTime: 0,
+    clientSideThrottleTime: decoded.throttleTime,
   }
 }
 
-const decodeResponses = decoder => ({
-  topic: decoder.readString(),
-  partitions: decoder.readArray(decodePartitions),
-})
-
-const decodePartitions = decoder => ({
-  partition: decoder.readInt32(),
-  errorCode: decoder.readInt16(),
-})
-
 module.exports = {
@@ -39,0 +28,0 @@ decode,

src/protocol/requests/offsetFetch/index.js

@@ -17,2 +17,7 @@ const versions = {
   },
+  4: ({ groupId, topics }) => {
+    const request = require('./v4/request')
+    const response = require('./v4/response')
+    return { request: request({ groupId, topics }), response }
+  },
 }
@@ -19,0 +24,0 @@

src/protocol/requests/produce/v6/response.js

@@ -1,3 +0,2 @@
-const Decoder = require('../../../decoder')
-const { parse: parseV5 } = require('../v5/response')
+const { parse, decode: decodeV5 } = require('../v5/response')
 
@@ -20,27 +19,9 @@ /**
 
-const partition = decoder => ({
-  partition: decoder.readInt32(),
-  errorCode: decoder.readInt16(),
-  baseOffset: decoder.readInt64().toString(),
-  logAppendTime: decoder.readInt64().toString(),
-  logStartOffset: decoder.readInt64().toString(),
-})
-
 const decode = async rawData => {
-  const decoder = new Decoder(rawData)
-  const topics = decoder.readArray(decoder => ({
-    topicName: decoder.readString(),
-    partitions: decoder.readArray(partition),
-  }))
+  const decoded = await decodeV5(rawData)
 
-  const throttleTime = decoder.readInt32()
-
-  // Report a `throttleTime` of 0: The broker will not have throttled
-  // this request, but if the `clientSideThrottleTime` is >0 then it
-  // expects us to do that -- and it will ignore requests.
-
   return {
-    topics,
+    ...decoded,
     throttleTime: 0,
-    clientSideThrottleTime: throttleTime,
+    clientSideThrottleTime: decoded.throttleTime,
   }
@@ -51,3 +32,3 @@ }
   decode,
-  parse: parseV5,
+  parse,
 }

src/protocol/requests/syncGroup/v2/response.js

@@ -1,2 +0,2 @@
-const { decode: decodeV1, parse: parseV1 } = require('../v1/response')
+const { parse, decode: decodeV1 } = require('../v1/response')
 
@@ -12,2 +12,3 @@ /**
  */
+
 const decode = async rawData => {
@@ -18,2 +19,3 @@ const decoded = await decodeV1(rawData)
     ...decoded,
+    throttleTime: 0,
     clientSideThrottleTime: decoded.throttleTime,
@@ -25,3 +27,3 @@ }
   decode,
-  parse: parseV1,
+  parse,
 }

src/protocol/requests/txnOffsetCommit/index.js

@@ -10,2 +10,10 @@ const versions = {
   },
+  1: ({ transactionalId, groupId, producerId, producerEpoch, topics }) => {
+    const request = require('./v1/request')
+    const response = require('./v1/response')
+    return {
+      request: request({ transactionalId, groupId, producerId, producerEpoch, topics }),
+      response,
+    }
+  },
 }
@@ -12,0 +20,0 @@

No alert changes

Improved metrics

Total package byte prevSize

670515

3.42%

Number of package files

378

7.39%

Lines of code

19873

3.43%

No dependency changes