WaterDrop

WaterDrop provides several methods for sending messages to Kafka, depending on whether you require synchronous or asynchronous behavior, and whether you are sending single messages or batches.

Synchronous Producing

Use produce_sync for a single message where you want to wait for the result, or produce_many_sync to send an array of messages synchronously.

Asynchronous Producing

Use produce_async for a single message to avoid blocking, or produce_many_async to send an array of messages asynchronously.

Transactions

You can wrap multiple produce_async calls within a transaction block to ensure atomicity.

# sync producing
producer.produce_sync(topic: 'my-topic', payload: 'my message')

# or for async
producer.produce_async(topic: 'my-topic', payload: 'my message')

# or in sync batches
producer.produce_many_sync(
  [
    { topic: 'my-topic', payload: 'my message'},
    { topic: 'my-topic', payload: 'my message'}
  ]
)

# and async batches
producer.produce_many_async(
  [
    { topic: 'my-topic', payload: 'my message'},
    { topic: 'my-topic', payload: 'my message'}
  ]
)

# transactions
producer.transaction do
  producer.produce_async(topic: 'my-topic', payload: 'my message')
  producer.produce_async(topic: 'my-topic', payload: 'my message')
end

The WaterDrop::Clients::Buffered client is a dummy implementation designed for use in test suites and specifications. Instead of sending messages to a real Kafka broker, it buffers them internally. This allows you to verify that specific messages were 'produced' by inspecting the client's state without requiring a running Kafka instance.

Key features:

• Message Buffering: Messages are stored in an internal array and can be retrieved by topic.
• Transactional Simulation: Supports begin_transaction, commit_transaction, and abort_transaction to simulate Kafka transactions.
• State Management: The reset method clears all buffered messages and transaction states, which is useful between test examples to prevent message leakage.

# Example of inspecting buffered messages in a test
client = WaterDrop::Clients::Buffered.new
client.produce(topic: 'my_topic', payload: 'hello')

# Retrieve messages for a specific topic
messages = client.messages_for('my_topic')
# => [{ topic: 'my_topic', payload: 'hello' }]

You can use the provided docker-compose.yml to spin up a local Kafka instance for development or testing with WaterDrop. This configuration uses the confluentinc/cp-kafka:8.3.0 image and sets up a single-node cluster with both broker and controller roles enabled.

services:
  kafka:
    container_name: kafka
    image: confluentinc/cp-kafka:8.3.0
    ports:
      - 9092:9092
    environment:
      CLUSTER_ID: kafka-docker-cluster-1
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092
      KAFKA_BROKER_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@127.0.0.1:9093
      ALLOW_PLAINTEXT_LISTENER: 'yes'
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_COORDINATOR_REBALANCE_PROTOCOLS: 'classic,consumer,share'
      KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR: 1
      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
      KAFKA_AUTHORIZER_CLASS_NAME: org.apache.kafka.metadata.authorizer.StandardAuthorizer

The WaterDrop::Instrumentation::Vendors::Datadog::MetricsListener allows you to subscribe to WaterDrop producer events and publish statistics to Datadog via a StatsD client.

To use it, you must first set up a dogstatsd-ruby client and pass it to the listener configuration. You can also customize the metric namespace and default tags (e.g., for host identification).

Configuration Options:

• namespace: The prefix for all published Datadog metrics. Defaults to "waterdrop".
• client: The Datadog/StatsD client instance (required).
• default_tags: An array of strings used as default tags for all metrics. Example: ["host:#{Socket.gethostname}"].
• rd_kafka_metrics: A list of RdKafkaMetric objects defining which rdkafka metrics to publish. This can be tuned to reduce noise.

# Example setup
WaterDrop::Instrumentation::Vendors::Datadog::MetricsListener.configure do |config|
  config.client = Datadog::Statsd.new('localhost', 8125)
  config.namespace = 'my_app.waterdrop'
  config.default_tags = ["env:production", "host:#{Socket.gethostname}"]
end

The WaterDrop::Instrumentation::LoggerListener is a default listener that hooks into WaterDrop instrumentation events to provide logging. It can be used standalone or as part of the Karafka framework listener.

It is designed to be non-intrusive; it can be removed or replaced without affecting the core WaterDrop flow.

When initializing, you can control the verbosity of the logs. By default, it logs message operations at the info level, but it also logs the full message content (payload and metadata) at the debug level. In environments like Rails development where the log level is often set to debug, this can result in extensive logs that flood the output. You can disable this behavior by setting log_messages: false.

# Initialize with a custom logger and control message content logging
listener = WaterDrop::Instrumentation::LoggerListener.new(Rails.logger, log_messages: false)