ApsaraMQ for Kafka:Send and receive messages with the Ruby SDK

SSL endpoint

Default endpoint

SSL endpoint

# frozen_string_literal: true

$LOAD_PATH.unshift(File.expand_path("../../lib", __FILE__))

require "kafka"

logger = Logger.new($stdout)
logger.level = Logger::INFO

# Replace the placeholders with your actual values.
brokers = "<your-broker-host1:port,your-broker-host2:port>"
topic = "<your-topic-name>"
username = "<your-sasl-username>"
password = "<your-sasl-password>"

kafka = Kafka.new(
    seed_brokers: brokers,
    client_id: "sasl-producer",
    logger: logger,
    # Path to the SSL root certificate file.
    ssl_ca_cert: File.read('./cert.pem'),
    sasl_plain_username: username,
    sasl_plain_password: password,
    )

producer = kafka.producer

# Read lines from stdin and send each line as a message.
begin
    $stdin.each_with_index do |line, index|

    producer.produce(line, topic: topic)

    producer.deliver_messages
end

ensure
    # Flush any remaining messages and release resources.
    producer.deliver_messages

    producer.shutdown
end

Default endpoint

# frozen_string_literal: true

$LOAD_PATH.unshift(File.expand_path("../../lib", __FILE__))

require "kafka"

logger = Logger.new($stdout)
logger.level = Logger::INFO

# Replace the placeholders with your actual values.
brokers = "<your-broker-host1:port,your-broker-host2:port>"
topic = "<your-topic-name>"

kafka = Kafka.new(
    seed_brokers: brokers,
    client_id: "simple-producer",
    logger: logger,
    )

producer = kafka.producer

# Read lines from stdin and send each line as a message.
begin
    $stdin.each_with_index do |line, index|

    producer.produce(line, topic: topic)

    producer.deliver_messages
end

ensure
    # Flush any remaining messages and release resources.
    producer.deliver_messages

    producer.shutdown
end

Run the producer:

ruby producer.ruby

Type each message on a new line and press Enter. The producer sends each line to the topic.

SSL endpoint

# frozen_string_literal: true

$LOAD_PATH.unshift(File.expand_path("../../lib", __FILE__))

require "kafka"

logger = Logger.new(STDOUT)
logger.level = Logger::INFO

# Replace the placeholders with your actual values.
brokers = "<your-broker-host1:port,your-broker-host2:port>"
topic = "<your-topic-name>"
username = "<your-sasl-username>"
password = "<your-sasl-password>"
consumerGroup = "<your-consumer-group-id>"

kafka = Kafka.new(
        seed_brokers: brokers,
        client_id: "sasl-consumer",
        socket_timeout: 20,
        logger: logger,
        # Path to the SSL root certificate file.
        ssl_ca_cert: File.read('./cert.pem'),
        sasl_plain_username: username,
        sasl_plain_password: password,
        )

consumer = kafka.consumer(group_id: consumerGroup)
consumer.subscribe(topic, start_from_beginning: false)

# Stop the consumer gracefully on TERM or INT signals.
# To stop the process, run: kill -s TERM <process-id>
trap("TERM") { consumer.stop }
trap("INT") { consumer.stop }

begin
    consumer.each_message(max_bytes: 64 * 1024) do |message|
    logger.info("Get message: #{message.value}")
    end
rescue Kafka::ProcessingError => e
    # If message processing fails, pause the affected partition
    # for 20 seconds before retrying. This prevents tight retry
    # loops that could overwhelm the broker.
    warn "Got error: #{e.cause}"
    consumer.pause(e.topic, e.partition, timeout: 20)

    retry
end

Default endpoint

# frozen_string_literal: true

$LOAD_PATH.unshift(File.expand_path("../../lib", __FILE__))

require "kafka"

logger = Logger.new(STDOUT)
logger.level = Logger::INFO

# Replace the placeholders with your actual values.
brokers = "<your-broker-host1:port,your-broker-host2:port>"
topic = "<your-topic-name>"
consumerGroup = "<your-consumer-group-id>"

kafka = Kafka.new(
        seed_brokers: brokers,
        client_id: "test",
        socket_timeout: 20,
        logger: logger,
        )

consumer = kafka.consumer(group_id: consumerGroup)
consumer.subscribe(topic, start_from_beginning: false)

# Stop the consumer gracefully on TERM or INT signals.
# To stop the process, run: kill -s TERM <process-id>
trap("TERM") { consumer.stop }
trap("INT") { consumer.stop }

begin
    consumer.each_message(max_bytes: 64 * 1024) do |message|
    logger.info("Get message: #{message.value}")
    end
rescue Kafka::ProcessingError => e
    # If message processing fails, pause the affected partition
    # for 20 seconds before retrying. This prevents tight retry
    # loops that could overwhelm the broker.
    warn "Got error: #{e.cause}"
    consumer.pause(e.topic, e.partition, timeout: 20)

    retry
end

Run the consumer:

ruby consumer.ruby

The consumer joins the specified consumer group and polls for new messages. Each message is logged to stdout. Press Ctrl+C to stop the consumer gracefully.