class Kafka::Broker

def commit_offsets(**options)
  request = Protocol::OffsetCommitRequest.new(**options)
  @connection.send_request(request)
end

def disconnect
  @connection.close
end

def fetch_messages(**options)
  request = Protocol::FetchRequest.new(**options)
  @connection.send_request(request)
end

def fetch_metadata(**options)
  request = Protocol::TopicMetadataRequest.new(**options)
  @connection.send_request(request)
end

def fetch_offsets(**options)
  request = Protocol::OffsetFetchRequest.new(**options)
  @connection.send_request(request)
end

def find_group_coordinator(**options)
  request = Protocol::GroupCoordinatorRequest.new(**options)
  @connection.send_request(request)
end

def heartbeat(**options)
  request = Protocol::HeartbeatRequest.new(**options)
  @connection.send_request(request)
end

def initialize(connection:, node_id: nil, logger:)
  @connection = connection
  @node_id = node_id
  @logger = logger
end

def join_group(**options)
  request = Protocol::JoinGroupRequest.new(**options)
  @connection.send_request(request)
end

def leave_group(**options)
  request = Protocol::LeaveGroupRequest.new(**options)
  @connection.send_request(request)
end

def list_offsets(**options)
  request = Protocol::ListOffsetRequest.new(**options)
  @connection.send_request(request)
end

def produce(**options)
  request = Protocol::ProduceRequest.new(**options)
  @connection.send_request(request)
end

def sasl_handshake(**options)
  request = Protocol::SaslHandshakeRequest(**options)
  @connection.send_request(request)
end

def sync_group(**options)
  request = Protocol::SyncGroupRequest.new(**options)
  @connection.send_request(request)
end

def to_s
  "#{@connection} (node_id=#{@node_id.inspect})"
end