Class: Kafka::Connection
- Inherits:
-
Object
- Object
- Kafka::Connection
- Defined in:
- lib/kafka/connection.rb
Overview
A connection to a single Kafka broker.
Usually you'll need a separate connection to each broker in a cluster, since most requests must be directed specifically to the broker that is currently leader for the set of topic partitions you want to produce to or consumer from.
Instrumentation
Connections emit a request.connection.kafka
notification on each request. The following
keys will be found in the payload:
:api
— the name of the API being invoked.:request_size
— the number of bytes in the request.:response_size
— the number of bytes in the response.
The notification also includes the duration of the request.
Constant Summary collapse
- SOCKET_TIMEOUT =
10
- CONNECT_TIMEOUT =
10
Instance Method Summary collapse
- #close ⇒ Object
-
#initialize(host:, port:, client_id:, logger:, connect_timeout: nil, socket_timeout: nil) ⇒ Connection
constructor
Opens a connection to a Kafka broker.
- #open? ⇒ Boolean
-
#send_request(request) ⇒ Object
Sends a request over the connection.
- #to_s ⇒ Object
Constructor Details
#initialize(host:, port:, client_id:, logger:, connect_timeout: nil, socket_timeout: nil) ⇒ Connection
Opens a connection to a Kafka broker.
45 46 47 48 49 50 51 |
# File 'lib/kafka/connection.rb', line 45 def initialize(host:, port:, client_id:, logger:, connect_timeout: nil, socket_timeout: nil) @host, @port, @client_id = host, port, client_id @logger = logger @connect_timeout = connect_timeout || CONNECT_TIMEOUT @socket_timeout = socket_timeout || SOCKET_TIMEOUT end |
Instance Method Details
#close ⇒ Object
61 62 63 64 65 66 67 |
# File 'lib/kafka/connection.rb', line 61 def close @logger.debug "Closing socket to #{to_s}" @socket.close if @socket @socket = nil end |
#open? ⇒ Boolean
57 58 59 |
# File 'lib/kafka/connection.rb', line 57 def open? !@socket.nil? end |
#send_request(request) ⇒ Object
Sends a request over the connection.
75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 |
# File 'lib/kafka/connection.rb', line 75 def send_request(request) # Default notification payload. notification = { api: Protocol.api_name(request.api_key), request_size: 0, response_size: 0, } Instrumentation.instrument("request.connection.kafka", notification) do open unless open? @correlation_id += 1 write_request(request, notification) response_class = request.response_class wait_for_response(response_class, notification) unless response_class.nil? end rescue Errno::EPIPE, Errno::ECONNRESET, Errno::ETIMEDOUT, EOFError => e close raise ConnectionError, "Connection error: #{e}" end |
#to_s ⇒ Object
53 54 55 |
# File 'lib/kafka/connection.rb', line 53 def to_s "#{@host}:#{@port}" end |