chore: sync CrowdFlower/ruby-kafka with zendesk/ruby-kafka

.circleci/config.yml

@@ -7,6 +7,7 @@ jobs:
           LOG_LEVEL: DEBUG
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec
       - run: bundle exec rubocop
@@ -40,6 +41,7 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec --profile --tag functional spec/functional
 
@@ -72,6 +74,7 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec --profile --tag functional spec/functional
 
@@ -104,6 +107,7 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec --profile --tag functional spec/functional
 
@@ -136,6 +140,7 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec --profile --tag functional spec/functional
 
@@ -168,6 +173,7 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec --profile --tag functional spec/functional
 
@@ -200,6 +206,7 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec --profile --tag functional spec/functional
 
@@ -232,6 +239,7 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec --profile --tag functional spec/functional
 
@@ -264,6 +272,7 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec --profile --tag functional spec/functional
 
@@ -296,6 +305,75 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
     steps:
       - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
+      - run: bundle install --path vendor/bundle
+      - run: bundle exec rspec --profile --tag functional spec/functional
+
+  kafka-2.6:
+    docker:
+      - image: circleci/ruby:2.5.1-node
+        environment:
+          LOG_LEVEL: DEBUG
+      - image: wurstmeister/zookeeper
+      - image: wurstmeister/kafka:2.13-2.6.0
+        environment:
+          KAFKA_ADVERTISED_HOST_NAME: localhost
+          KAFKA_ADVERTISED_PORT: 9092
+          KAFKA_PORT: 9092
+          KAFKA_ZOOKEEPER_CONNECT: localhost:2181
+          KAFKA_DELETE_TOPIC_ENABLE: true
+      - image: wurstmeister/kafka:2.13-2.6.0
+        environment:
+          KAFKA_ADVERTISED_HOST_NAME: localhost
+          KAFKA_ADVERTISED_PORT: 9093
+          KAFKA_PORT: 9093
+          KAFKA_ZOOKEEPER_CONNECT: localhost:2181
+          KAFKA_DELETE_TOPIC_ENABLE: true
+      - image: wurstmeister/kafka:2.13-2.6.0
+        environment:
+          KAFKA_ADVERTISED_HOST_NAME: localhost
+          KAFKA_ADVERTISED_PORT: 9094
+          KAFKA_PORT: 9094
+          KAFKA_ZOOKEEPER_CONNECT: localhost:2181
+          KAFKA_DELETE_TOPIC_ENABLE: true
+    steps:
+      - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
+      - run: bundle install --path vendor/bundle
+      - run: bundle exec rspec --profile --tag functional spec/functional
+
+  kafka-2.7:
+    docker:
+      - image: circleci/ruby:2.5.1-node
+        environment:
+          LOG_LEVEL: DEBUG
+      - image: bitnami/zookeeper
+        environment:
+          ALLOW_ANONYMOUS_LOGIN: yes
+      - image: bitnami/kafka:2.7.0
+        environment:
+          ALLOW_PLAINTEXT_LISTENER: yes
+          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
+          KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
+          KAFKA_CFG_ZOOKEEPER_CONNECT: localhost:2181
+          KAFKA_DELETE_TOPIC_ENABLE: true
+      - image: bitnami/kafka:2.7.0
+        environment:
+          ALLOW_PLAINTEXT_LISTENER: yes
+          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9093
+          KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9093
+          KAFKA_CFG_ZOOKEEPER_CONNECT: localhost:2181
+          KAFKA_DELETE_TOPIC_ENABLE: true
+      - image: bitnami/kafka:2.7.0
+        environment:
+          ALLOW_PLAINTEXT_LISTENER: yes
+          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9094
+          KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9094
+          KAFKA_CFG_ZOOKEEPER_CONNECT: localhost:2181
+          KAFKA_DELETE_TOPIC_ENABLE: true
+    steps:
+      - checkout
+      - run: sudo apt-get update && sudo apt-get install -y cmake # For installing snappy
       - run: bundle install --path vendor/bundle
       - run: bundle exec rspec --profile --tag functional spec/functional
 
@@ -313,3 +391,5 @@ workflows:
       - kafka-2.3
       - kafka-2.4
       - kafka-2.5
+      - kafka-2.6
+      - kafka-2.7

CHANGELOG.md

@@ -4,6 +4,21 @@ Changes and additions to the library will be listed here.
 
 ## Unreleased
 
+## 1.5.0
+- Add support for AWS IAM Authentication to an MSK cluster (#907).
+- Added session token to the IAM mechanism; necessary for auth via temporary credentials (#937)
+
+## 1.4.0
+
+- Refresh a stale cluster's metadata if necessary on `Kafka::Client#deliver_message` (#901).
+- Fix `Kafka::TransactionManager#send_offsets_to_txn` (#866).
+- Add support for `murmur2` based partitioning.
+- Add `resolve_seed_brokers` option to support seed brokers' hostname with multiple addresses (#877).
+- Handle SyncGroup responses with a non-zero error and no assignments (#896).
+- Add support for non-identical topic subscriptions within the same consumer group (#525 / #764).
+
+## 1.3.0
+
 - Support custom assignment strategy (#846).
 - Improved Exceptions in TransactionManager (#862).
 

README.md

@@ -129,6 +129,16 @@ Or install it yourself as:
     <td>Limited support</td>
     <td>Limited support</td>
   </tr>
+  <tr>
+    <th>Kafka 2.6</th>
+    <td>Limited support</td>
+    <td>Limited support</td>
+  </tr>
+  <tr>
+    <th>Kafka 2.7</th>
+    <td>Limited support</td>
+    <td>Limited support</td>
+  </tr>
 </table>
 
 This library is targeting Kafka 0.9 with the v0.4.x series and Kafka 0.10 with the v0.5.x series. There's limited support for Kafka 0.8, and things should work with Kafka 0.11, although there may be performance issues due to changes in the protocol.
@@ -144,6 +154,8 @@ This library is targeting Kafka 0.9 with the v0.4.x series and Kafka 0.10 with t
 - **Kafka 2.3:** Everything that works with Kafka 2.2 should still work, but so far no features specific to Kafka 2.3 have been added.
 - **Kafka 2.4:** Everything that works with Kafka 2.3 should still work, but so far no features specific to Kafka 2.4 have been added.
 - **Kafka 2.5:** Everything that works with Kafka 2.4 should still work, but so far no features specific to Kafka 2.5 have been added.
+- **Kafka 2.6:** Everything that works with Kafka 2.5 should still work, but so far no features specific to Kafka 2.6 have been added.
+- **Kafka 2.7:** Everything that works with Kafka 2.6 should still work, but so far no features specific to Kafka 2.7 have been added.
 
 This library requires Ruby 2.1 or higher.
 
@@ -164,6 +176,12 @@ require "kafka"
 kafka = Kafka.new(["kafka1:9092", "kafka2:9092"], client_id: "my-application")
 ```
 
+You can also use a hostname with seed brokers' IP addresses:
+
+```ruby
+kafka = Kafka.new("seed-brokers:9092", client_id: "my-application", resolve_seed_brokers: true)
+```
+
 ### Producing Messages to Kafka
 
 The simplest way to write a message to a Kafka topic is to call `#deliver_message`:
@@ -370,6 +388,16 @@ partitioner = -> (partition_count, message) { ... }
 Kafka.new(partitioner: partitioner, ...)
 ```
 
+##### Supported partitioning schemes
+
+In order for semantic partitioning to work a `partition_key` must map to the same partition number every time. The general approach, and the one used by this library, is to hash the key and mod it by the number of partitions. There are many different algorithms that can be used to calculate a hash. By default `crc32` is used. `murmur2` is also supported for compatibility with Java based Kafka producers.
+
+To use `murmur2` hashing pass it as an argument to `Partitioner`. For example:
+
+```ruby
+Kafka.new(partitioner: Kafka::Partitioner.new(hash_function: :murmur2))
+```
+
 #### Buffering and Error Handling
 
 The producer is designed for resilience in the face of temporary network errors, Kafka broker failovers, and other issues that prevent the client from writing messages to the destination topics. It does this by employing local, in-memory buffers. Only when messages are acknowledged by a Kafka broker will they be removed from the buffer.
@@ -1093,6 +1121,20 @@ kafka = Kafka.new(
 )
 ```
 
+##### AWS MSK (IAM)
+In order to authenticate using IAM w/ an AWS MSK cluster, set your access key, secret key, and region when initializing the Kafka client:
+
+```ruby
+k = Kafka.new(
+  ["kafka1:9092"],
+  sasl_aws_msk_iam_access_key_id: 'iam_access_key',
+  sasl_aws_msk_iam_secret_key_id: 'iam_secret_key',
+  sasl_aws_msk_iam_aws_region: 'us-west-2',
+  ssl_ca_certs_from_system: true,
+  # ...
+)
+```
+
 ##### PLAIN
 In order to authenticate using PLAIN, you must set your username and password when initializing the Kafka client:
 

lib/kafka/async_producer.rb

@@ -212,31 +212,45 @@ def run
         @logger.push_tags(@producer.to_s)
         @logger.info "Starting async producer in the background..."
 
+        do_loop
+      rescue Exception => e
+        @logger.error "Unexpected Kafka error #{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"
+        @logger.error "Async producer crashed!"
+      ensure
+        @producer.shutdown
+        @logger.pop_tags
+      end
+
+      private
+
+      def do_loop
         loop do
-          operation, payload = @queue.pop
-
-          case operation
-          when :produce
-            produce(payload[0], **payload[1])
-            deliver_messages if threshold_reached?
-          when :deliver_messages
-            deliver_messages
-          when :shutdown
-            begin
-              # Deliver any pending messages first.
-              @producer.deliver_messages
-            rescue Error => e
-              @logger.error("Failed to deliver messages during shutdown: #{e.message}")
-
-              @instrumenter.instrument("drop_messages.async_producer", {
-                message_count: @producer.buffer_size + @queue.size,
-              })
+          begin
+            operation, payload = @queue.pop
+
+            case operation
+            when :produce
+              produce(payload[0], **payload[1])
+              deliver_messages if threshold_reached?
+            when :deliver_messages
+              deliver_messages
+            when :shutdown
+              begin
+                # Deliver any pending messages first.
+                @producer.deliver_messages
+              rescue Error => e
+                @logger.error("Failed to deliver messages during shutdown: #{e.message}")
+
+                @instrumenter.instrument("drop_messages.async_producer", {
+                  message_count: @producer.buffer_size + @queue.size,
+                })
+              end
+
+              # Stop the run loop.
+              break
+            else
+              raise "Unknown operation #{operation.inspect}"
             end
-
-            # Stop the run loop.
-            break
-          else
-            raise "Unknown operation #{operation.inspect}"
           end
         end
       rescue Kafka::Error => e
@@ -245,16 +259,8 @@ def run
 
         sleep 10
         retry
-      rescue Exception => e
-        @logger.error "Unexpected Kafka error #{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"
-        @logger.error "Async producer crashed!"
-      ensure
-        @producer.shutdown
-        @logger.pop_tags
       end
 
-      private
-
       def produce(value, **kwargs)
         retries = 0
         begin