Merge branch 'dev_producer_testing_and_fixes' into dev_early_access_development_branch

Author: milindl

MIGRATION.md

@@ -37,19 +37,18 @@
   * `clientId`: string for identifying this client.
   * **`connectionTimeout`** and **`authenticationTimeout`**:
           These timeouts (specified in milliseconds) are not enforced individually. Instead, the sum of these values is
-          enforced. The default value of the sum is 30000. It corresponds to librdkafka's `socket.connection.setup.timeout.ms`.
+          enforced. The default value of the sum is 11000, same as for KafkaJS.
+          It corresponds to librdkafka's `socket.connection.setup.timeout.ms`.
   * **`reauthenticationThreshold`**: no longer checked, librdkafka handles reauthentication on its own.
   * **`requestTimeout`**: number of milliseconds for a network request to timeout. The default value has been changed to 60000. It now corresponds to librdkafka's `socket.timeout.ms`.
   * **`enforceRequestTimeout`**: if this is set to false, `requestTimeout` is set to 5 minutes. The timeout cannot be disabled completely.
   * **`retry`** is partially supported. It must be an object, with the following (optional) properties
-    - `maxRetryTime`: maximum time to backoff a retry, in milliseconds. Corresponds to librdkafka's `retry.backoff.max.ms`. The default is 1000.
-    - `initialRetryTime`: minimum time to backoff a retry, in milliseconds. Corresponds to librdkafka's `retry.backoff.ms`. The default is 100.
-    - `retries`: maximum number of retries, *only* applicable to Produce messages. However, it's recommended to keep this unset.
-                 Librdkafka handles the number of retries, and rather than capping the number of retries, caps the total time spent
-                 while sending the message, controlled by `message.timeout.ms`.
-    - `factor` and `multiplier` cannot be changed from their defaults of 0.2 and 2.
-  * **`restartOnFailure`**: this cannot be changed, and will always be true (the consumer recovers from errors on its own).
-  * `logLevel` is mapped to the syslog(3) levels supported by librdkafka. `LOG_NOTHING` is not YET supported, as some panic situations are still logged.
+    - `maxRetryTime`: maximum time to backoff a retry, in milliseconds. Corresponds to librdkafka's `retry.backoff.max.ms`.
+    - `initialRetryTime`: minimum time to backoff a retry, in milliseconds. Corresponds to librdkafka's `retry.backoff.ms`.
+    - `retries`: maximum number of retries, *only* applicable to Produce messages.
+    - **`factor`** and **`multiplier`** cannot be changed from their defaults of 0.2 and 2.
+    - **`restartOnFailure`**: this cannot be changed, and will always be true (the consumer recovers from errors on its own).
+  * `logLevel` is mapped automatically to the syslog(3) levels supported by librdkafka. `LOG_NOTHING` is not YET supported, as some panic situations are still logged.
   * **`socketFactory`** is no longer supported.
 
 #### Error Handling
@@ -119,9 +118,9 @@
   There are several changes in the common configuration. Each config property is discussed.
   If there needs to be any change, the property is highlighted.
 
-  * **`createPartitioner`**: this is not supported (YET). For behaviour identical to the Java client (the DefaultPartitioner),
-                             use the `rdKafka` block, and set the property `partitioner` to `murmur2_random`. This is critical
-                             when planning to produce to topics where messages with certain keys have been produced already.
+  * **`createPartitioner`**: this is not supported yet. The default behaviour is identical to the DefaultPartitioner, and compatible with Java client's default
+                            partitioner.
+                            This corresponds to the librdkafka property `partitioner` and the value `murmur2_random`.
   * **`retry`**: See the section for retry above. The producer config `retry` takes precedence over the common config `retry`.
   * `metadataMaxAge`: Time in milliseconds after which to refresh metadata for known topics. The default value remains 5min. This
                       corresponds to the librdkafka property `topic.metadata.refresh.interval.ms` (and not `metadata.max.age.ms`).
@@ -131,7 +130,7 @@
                            Only applicable when `transactionalId` is set to true.
   * `idempotent`: if set to true, ensures that messages are delivered exactly once and in order. False by default.
                   In case this is set to true, certain constraints must be respected for other properties, `maxInFlightRequests <= 5`, `retry.retries >= 0`.
-  * **`maxInFlightRequests`**: Maximum number of in-flight requests *per broker connection*. If not set, a very high limit is used.
+  * **`maxInFlightRequests`**: Maximum number of in-flight requests *per broker connection*. If not set, it is practically unbounded (same as KafkaJS).
   * `transactionalId`: if set, turns this into a transactional producer with this identifier. This also automatically sets `idempotent` to true.
   * An `rdKafka` block can be added to the config. It allows directly setting librdkafka properties.
     If you are starting to make the configuration anew, it is best to specify properties using
@@ -140,7 +139,8 @@
 #### Semantic and Per-Method Changes
 
 * Changes to `send`:
-  * `acks`, `compression` and `timeout` are not set on a per-send basis. Rather, they must be configured in the configuration.
+  * `acks`, `compression` and `timeout` are not set on a per-send basis. Rather, they must be configured in the top-level configuration.
+    Additionally, there are several more compression types available by default besides GZIP.
     Before:
     ```javascript
     const kafka = new Kafka({/* ... */});
@@ -160,13 +160,9 @@
     ```javascript
     const kafka = new Kafka({/* ... */});
     const producer = kafka.producer({
-      rdKafka: {
-        topicConfig: {
-          "acks": "1",
-          "compression.codec": "gzip",
-          "message.timeout.ms": "30000",
-        },
-      }
+      acks: 1,
+      compression: CompressionTypes.GZIP|CompressionTypes.SNAPPY|CompressionTypes.LZ4|CompressionTypes.ZSTD|CompressionTypes.NONE,
+      timeout: 30000,
     });
     await producer.connect();
 
@@ -191,27 +187,28 @@
   If there needs to be any change, the property is highlighted. The change could be a change in
   the default values, some added/missing features, or a change in semantics.
 
-  * **`partitionAssigners`**: The **default value** of this is changed to `[PartitionAssigners.range,PartitionAssigners.roundRobin]`. Support for range, roundRobin and cooperativeSticky
-                              partition assignors is provided. The cooperative assignor cannot be used along with the other two, and there
-                              is no support for custom assignors. An alias for these properties is also made available, `partitionAssignors` and `PartitionAssignors` to maintain
-                              parlance with the Java client's terminology.
-  * **`sessionTimeout`**: If no heartbeats are received by the broker for a group member within the session timeout, the broker will remove the consumer from
-                         the group and trigger a rebalance. The **default value** is changed to 45000.
+  * `partitionAssigners`: Support for range and roundRobin assignors is provided. Custom assignors are not supported.
+                          The default value of this remains `[PartitionAssigners.roundRobin]`.
+                          Support for cooperative-sticky assignor will be added soon.
+                          An alias for these properties is also made available, `partitionAssignors` and `PartitionAssignors` to maintain
+                          parlance with the Java client's terminology.
+  * `sessionTimeout`: If no heartbeats are received by the broker for a group member within the session timeout, the broker will remove the consumer from
+                         the group and trigger a rebalance.
   * **`rebalanceTimeout`**: The maximum allowed time for each member to join the group once a rebalance has begun. The **default value** is changed to 300000.
                             Note, before changing: setting this value *also* changes the max poll interval. Message processing in `eachMessage` must not take more than this time.
   * `heartbeatInterval`: The expected time in milliseconds between heartbeats to the consumer coordinator. The default value remains 3000.
   * `metadataMaxAge`: Time in milliseconds after which to refresh metadata for known topics. The default value remains 5min. This
                       corresponds to the librdkafka property `topic.metadata.refresh.interval.ms` (and not `metadata.max.age.ms`).
-  * **`allowAutoTopicCreation`**: determines if a topic should be created if it doesn't exist while producing. The **default value** is changed to false.
+  * `allowAutoTopicCreation`: determines if a topic should be created if it doesn't exist while producing.
   * **`maxBytesPerPartition`**: determines how many bytes can be fetched in one request from a single partition. The default value remains 1048576.
-                                There is a slight change in semantics, this size grows dynamically if a single message larger than this is encountered,
+                                There is a change in semantics, this size grows dynamically if a single message larger than this is encountered,
                                 and the client does not get stuck.
   * `minBytes`: Minimum number of bytes the broker responds with (or wait until `maxWaitTimeInMs`). The default remains 1.
-  * **`maxBytes`**: Maximum number of bytes the broker responds with. The **default value** is changed to 52428800 (50MB).
-  * **`maxWaitTimeInMs`**: Maximum time in milliseconds the broker waits for the `minBytes` to be fulfilled. The **default value** is changed to 500.
+  * `maxBytes`: Maximum number of bytes the broker responds with.
+  * `maxWaitTimeInMs`: Maximum time in milliseconds the broker waits for the `minBytes` to be fulfilled.
   * **`retry`**: See the section for retry above. The consumer config `retry` takes precedence over the common config `retry`.
   * `readUncommitted`: if true, consumer will read transactional messages which have not been committed. The default value remains false.
-  * **`maxInFlightRequests`**: Maximum number of in-flight requests *per broker connection*. If not set, a very high limit is used.
+  * **`maxInFlightRequests`**: Maximum number of in-flight requests *per broker connection*. If not set, it is practically unbounded (same as KafkaJS).
   * `rackId`: Can be set to an arbitrary string which will be used for fetch-from-follower if set up on the cluster.
   * An `rdKafka` block can be added to the config. It allows directly setting librdkafka properties.
     If you are starting to make the configuration anew, it is best to specify properties using
@@ -225,10 +222,9 @@
   * Subscribe must be called after `connect`.
   * An optional parameter, `replace` is provided. If set to true, the current subscription is replaced with the new one. If set to false, the new subscription is added to the current one.
     The default value is false.
-  * While passing a list of topics to `subscribe`, the `fromBeginning` property is not supported. Instead, the property `auto.offset.reset` needs to be used.
+  * While passing a list of topics to `subscribe`, the `fromBeginning` is not set on a per-subscribe basis. Rather, it must be configured in the top-level configuration.
    Before:
     ```javascript
-      const kafka = new Kafka({ /* ... */ });
       const consumer = kafka.consumer({
         groupId: 'test-group',
       });
@@ -237,23 +233,17 @@
     ```
    After:
     ```javascript
-      const kafka = new Kafka({ /* ... */ });
       const consumer = kafka.consumer({
         groupId: 'test-group',
-        rdKafka: {
-          topicConfig: {
-            'auto.offset.reset': 'earliest',
-          },
-        }
+        fromBeginning: true,
       });
       await consumer.connect();
       await consumer.subscribe({ topics: ["topic"] });
     ```
 
- * For auto-committing using a consumer, the properties on `run` are no longer used. Instead, corresponding rdKafka properties must be set.
-    * `autoCommit` corresponds to `enable.auto.commit`.
-    * `autoCommitInterval` corresponds to `auto.commit.interval.ms`.
-    * `autoCommitThreshold` is no longer supported.
+ * For auto-committing using a consumer, the properties `autoCommit` and `autoCommitInterval` on `run` are not set on a per-subscribe basis.
+   Rather, they must be configured in the top-level configuration.
+   `autoCommitThreshold` is not supported.
 
     Before:
     ```javascript
@@ -273,12 +263,8 @@
       const kafka = new Kafka({ /* ... */ });
       const consumer = kafka.consumer({
         /* ... */,
-        rdKafka: {
-          globalConfig: {
-            "enable.auto.commit": "true",
-            "auto.commit.interval.ms": "5000",
-          }
-        },
+        autoCommit: true,
+        autoCommitThreshold: 5000,
       });
       await consumer.connect();
       await consumer.subscribe({ topics: ["topic"] });
@@ -287,18 +273,17 @@
       });
     ```
 
+  * The `partitionsConsumedConcurrently` property is not supported at the moment.
+  * The `eachBatch` method is not supported.
   * For the `eachMessage` method while running the consumer:
     * The `heartbeat()` no longer needs to be called. Heartbeats are automatically managed by librdkafka.
-    * The `partitionsConsumedConcurrently` property is not supported (YET).
-  * The `eachBatch` method is not supported.
-  * `commitOffsets` does not (YET) support sending metadata for topic partitions being committed.
+  * `commitOffsets` does not yet support sending metadata for topic partitions being committed.
   * `paused()` is supported without any changes.
-  * Custom partition assignors are not supported.
   * Changes to `seek`:
     * The restriction to call seek only after `run` is removed. It can be called any time.
-    * Rather than the `autoCommit` property of `run` deciding if the offset is committed, the librdkafka property `enable.auto.commit` of the consumer config is used.
   * `pause` and `resume` MUST be called after the consumer group is joined. In practice, this means it can be called whenever `consumer.assignment()` has a non-zero size, or within the `eachMessage`
     callback.
+  * `stop` is not yet supported, and the user must disconnect the consumer.
 
 ### Admin Client