search

doujiang24 / lua-resty-kafka

Lua kafka client driver for the Openresty based on the cosocket API
BSD 3-Clause "New" or "Revised" License
801 stars 274 forks source link

readme

Name

lua-resty-kafka - Lua kafka client driver for the ngx_lua based on the cosocket API

Table of Contents

Status

This library is still under early development and is still experimental.

Description

This Lua library is a Kafka client driver for the ngx_lua nginx module:

http://wiki.nginx.org/HttpLuaModule

This Lua library takes advantage of ngx_lua's cosocket API, which ensures 100% nonblocking behavior.

Note that at least ngx_lua 0.9.3 or openresty 1.4.3.7 is required, and unfortunately only LuaJIT supported (--with-luajit).

Note for ssl connections at least ngx_lua 0.9.11 or openresty 1.7.4.1 is required, and unfortunately only LuaJIT supported (--with-luajit).

Synopsis

    lua_package_path "/path/to/lua-resty-kafka/lib/?.lua;;";

    server {
        location /test {
            content_by_lua '
                local cjson = require "cjson"
                local client = require "resty.kafka.client"
                local producer = require "resty.kafka.producer"

                local broker_list = {
                    {
                        host = "127.0.0.1",
                        port = 9092,

                        -- optional auth
                        sasl_config = {
                            mechanism = "PLAIN",
                            user = "USERNAME",
                            password = "PASSWORD",
                        },
                    },
                }

                local key = "key"
                local message = "halo world"

                -- usually we do not use this library directly
                local cli = client:new(broker_list)
                local brokers, partitions = cli:fetch_metadata("test")
                if not brokers then
                    ngx.say("fetch_metadata failed, err:", partitions)
                end
                ngx.say("brokers: ", cjson.encode(brokers), "; partitions: ", cjson.encode(partitions))

                -- sync producer_type
                local p = producer:new(broker_list)

                local offset, err = p:send("test", key, message)
                if not offset then
                    ngx.say("send err:", err)
                    return
                end
                ngx.say("send success, offset: ", tonumber(offset))

                -- this is async producer_type and bp will be reused in the whole nginx worker
                local bp = producer:new(broker_list, { producer_type = "async" })

                local ok, err = bp:send("test", key, message)
                if not ok then
                    ngx.say("send err:", err)
                    return
                end

                ngx.say("send success, ok:", ok)
            ';
        }
    }

Back to TOC

Modules

resty.kafka.client

To load this module, just do this

    local client = require "resty.kafka.client"

Back to TOC

Methods

new

syntax: c = client:new(broker_list, client_config)

The broker_list is a list of broker, like the below

[
    {
        "host": "127.0.0.1",
        "port": 9092,

        // optional auth
        "sasl_config": {
            //support mechanism: PLAIN、SCRAM-SHA-256、SCRAM-SHA-512
            "mechanism": "PLAIN",
            "user": "USERNAME",
            "password": "PASSWORD"
        }
    }
]

An optional client_config table can be specified. The following options are as follows:

client config

Back to TOC

fetch_metadata

syntax: brokers, partitions = c:fetch_metadata(topic)

In case of success, return the all brokers and partitions of the topic. In case of errors, returns nil with a string describing the error.

Back to TOC

refresh

syntax: brokers, partitions = c:refresh()

This will refresh the metadata of all topics which have been fetched by fetch_metadata. In case of success, return all brokers and all partitions of all topics. In case of errors, returns nil with a string describing the error.

Back to TOC

choose_api_version

syntax: api_version = c:choose_api_version(api_key, min_version, max_version)

This helps the client to select the correct version of the api_key corresponding to the API.

When min_version and max_version are provided, it will act as a limit and the selected versions in the return value will not exceed their limits no matter how high or low the broker supports the API version. When they are not provided, it will follow the range of versions supported by the broker.

Tip: The version selection strategy is to choose the maximum version within the allowed range.

Back to TOC

resty.kafka.producer

To load this module, just do this

    local producer = require "resty.kafka.producer"

Back to TOC

Methods

new

syntax: p = producer:new(broker_list, producer_config?, cluster_name?)

It's recommend to use async producer_type.

broker_list is the same as in client

An optional options table can be specified. The following options are as follows:

socket_timeout, keepalive_timeout, keepalive_size, refresh_interval, ssl, ssl_verify are the same as in client_config

producer config, most like in http://kafka.apache.org/documentation.html#producerconfigs

buffer config ( only work producer_type = "async" )

Not support compression now.

The third optional cluster_name specifies the name of the cluster, default 1 (yeah, it's number). You can Specifies different names when you have two or more kafka clusters. And this only works with async producer_type.

Back to TOC

send

syntax: ok, err = p:send(topic, key, message)

  1. In sync model

    In case of success, returns the offset ( cdata: LL ) of the current broker and partition. In case of errors, returns nil with a string describing the error.

  2. In async model

    The message will write to the buffer first. It will send to the kafka server when the buffer exceed the batch_num, or every flush_time flush the buffer.

    It case of success, returns true. In case of errors, returns nil with a string describing the error (buffer overflow).

Back to TOC

offset

syntax: sum, details = p:offset()

Return the sum of all the topic-partition offset (return by the ProduceRequest api);
and the details of each topic-partition

Back to TOC

flush

syntax: ok = p:flush()

Always return true.

Back to TOC

resty.kafka.basic-consumer

To load this module, just do this

    local bconsumer = require "resty.kafka.basic-consumer"

This module is a minimalist implementation of a consumer, providing the list_offset API for querying by time or getting the start and end offset and the fetch API for getting messages in a topic.

In a single call, only the information of a single partition in a single topic can be fetched, and batch fetching is not supported for now. The basic consumer does not support the consumer group related API, so you need to fetch the message after getting the offset through the list_offset API, or your service can manage the offset itself.

Back to TOC

Methods

new

syntax: c = bconsumer:new(broker_list, client_config)

The broker_list is a list of broker, like the below

[
    {
        "host": "127.0.0.1",
        "port": 9092,

        // optional auth
        "sasl_config": {
            "mechanism": "PLAIN",
            "user": "USERNAME",
            "password": "PASSWORD"
        }
    }
]

An optional client_config table can be specified. The following options are as follows:

client config

Back to TOC

list_offset

syntax: offset, err = c:list_offset(topic, partition, timestamp)

The parameter timestamp can be a UNIX timestamp or a constant defined in resty.kafka.protocol.consumer, LIST_OFFSET_TIMESTAMP_LAST, LIST_OFFSET_TIMESTAMP_FIRST, LIST_OFFSET_TIMESTAMP_MAX, used to get the initial and latest offsets, etc., semantics with the ListOffsets API in Apache Kafka. See: https://kafka.apache.org/protocol.html#The_Messages_ListOffsets

In case of success, return the offset of the specified case. In case of errors, returns nil with a string describing the error.

Back to TOC

fetch

syntax: result, err = c:fetch(topic, partition, offset)

In case of success, return the following result of the specified case. In case of errors, returns nil with a string describing the error.

The result will contain more information such as the messages:

Back to TOC

Errors

When you call the modules provided in this library, you may get some errors. Depending on the source, they can be divided into the following categories.

Installation

You need to configure the lua_package_path directive to add the path of your lua-resty-kafka source tree to ngx_lua's LUA_PATH search path, as in

    # nginx.conf
    http {
        lua_package_path "/path/to/lua-resty-kafka/lib/?.lua;;";
        ...
    }

Ensure that the system account running your Nginx ''worker'' proceses have enough permission to read the .lua file.

Back to TOC

TODO

  1. Fetch API
  2. Offset API
  3. Offset Commit/Fetch API

Back to TOC

Author

Dejiang Zhu (doujiang24) doujiang24@gmail.com.

Back to TOC

Copyright and License

This module is licensed under the BSD license.

Copyright (C) 2014-2020, by Dejiang Zhu (doujiang24) doujiang24@gmail.com.

All rights reserved.

Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

Back to TOC

See Also

Back to TOC