kashka v0.1.0 Kashka.Kafka View Source

Module to make direct requests to Kafka Rest Proxy. Use Confluent REST Proxy API Reference to figure out how to use

Link to this section Summary

Types

binary_records()

create_consumer_params()

http_error()

json_records()

partition()

records_query_params()

Functions

assignments(conn)

Get the list of partitions currently manually assigned to this consumer. See Confluent REST Proxy API Reference

close(conn)

Close HTTP connection.

commit(conn, offsets_or_records \\ nil)

Commit a list of offsets for the consumer. See Confluent REST Proxy API Reference

create_consumer(conn, consumer_group, opts)

Create a new consumer instance in the consumer group. See Confluent REST Proxy API Reference

delete_consumer(conn)

Destroy the consumer instance. See Confluent REST Proxy API Reference

delete_consumer(conn, group, name)

Destroy the consumer instance. See Confluent REST Proxy API Reference. This function mostly used in tests when there is only one kafka rest proxy instance.

get_records(conn, opts, format \\ :json)

Produce messages to a topic. See Confluent REST Proxy API Reference

move_to_existing_consumer(conn, group, name)

Build connection to exsisting consumer. This function mostly used in tests when there is only one kafka rest proxy instance

offsets(conn, partitions)

Requests offsets for consumer. See Confluent REST Proxy API Reference

positions_end(conn, partitions)

Seek to the last offset for each of the given partitions. See Confluent REST Proxy API Reference

produce(conn, topic, records, format \\ :json)

Produce messages to a topic. See Confluent REST Proxy API Reference

subscribe(conn, topics)

Subscribe to the given list of topics. See Confluent REST Proxy API Reference

topics(conn)

Requests all topics list See Confluent REST Proxy API Reference

Link to this section Types

binary_records()

binary_records() :: %{
  :key => String.t(),
  :value => String.t(),
  optional(:partition) => integer()
}

create_consumer_params()

create_consumer_params() :: %{
  optional(:name) => String.t(),
  optional(:format) => :binary | :json,
  optional(:"auto.offset.reset") => :earliest | :latest,
  optional(:"auto.commit.enable") => true | false,
  optional(:"fetch.min.bytes") => integer(),
  optional(:"consumer.request.timeout.ms") => integer()
}

http_error()

http_error() :: {:error, :http, code :: non_neg_integer(), iodata()}

json_records()

json_records() :: %{
  :key => String.t(),
  :value => any(),
  optional(:partition) => integer()
}

partition()

partition() :: %{topic: String.t() | atom(), partition: integer()}

records_query_params()

records_query_params() :: %{
  optional(:timeout) => integer(),
  optional(:max_bytes) => integer()
}

Link to this section Functions

assignments(conn)

assignments(Kashka.Http.t()) :: {:ok, Kashka.Http.t(), [any()]} | http_error()

Get the list of partitions currently manually assigned to this consumer. See Confluent REST Proxy API Reference

Parameters

conn: connection or url for created consumer

close(conn)

close(Kashka.Http.t()) :: :ok

Close HTTP connection.

commit(conn, offsets_or_records \\ nil)

commit(Kashka.Http.conn(), [map()] | nil) ::
  {:ok, Kashka.Http.t()} | http_error()

Commit a list of offsets for the consumer. See Confluent REST Proxy API Reference

Parameters

conn: connection or url for created consumer
offsets_or_records: offsets list or records returned from get_records/3

create_consumer(conn, consumer_group, opts)

create_consumer(Kashka.Http.t(), String.t(), create_consumer_params()) ::
  {:ok, Kashka.Http.t(), map()} | {:error, :exists} | http_error()

Create a new consumer instance in the consumer group. See Confluent REST Proxy API Reference

Parameters

conn: connection or url for created consumer
consumer_group: the name of the consumer group to join
opts: consumer settings

Returns map with "base_uri" and "instance_id" fields. For example:

  %{
    "instance_id" => "my_consumer",
    "base_uri" => "http://proxy-instance.kafkaproxy.example.com/consumers/testgroup/instances/my_consumer"
  }

delete_consumer(conn)

delete_consumer(Kashka.Http.t()) :: {:ok, Kashka.Http.t()} | http_error()

Destroy the consumer instance. See Confluent REST Proxy API Reference

Parameters

conn: connection or url for created consumer

delete_consumer(conn, group, name)

delete_consumer(Kashka.Http.t(), String.t(), String.t()) ::
  {:ok, Kashka.Http.t()} | http_error()

Destroy the consumer instance. See Confluent REST Proxy API Reference. This function mostly used in tests when there is only one kafka rest proxy instance.

Parameters

conn: connection or url for rest proxy instance
group: consumer group name
name: consumer instance_id

get_records(conn, opts, format \\ :json)

get_records(Kashka.Http.conn(), records_query_params(), :json | :binary) ::
  {:ok, Kashka.Http.t(), [map()]} | http_error()

Produce messages to a topic. See Confluent REST Proxy API Reference

Parameters

conn: connection or url for created consumer
opts: query Parameters
format: created consumer records format: json or binary

move_to_existing_consumer(conn, group, name)

move_to_existing_consumer(Kashka.Http.t(), String.t(), String.t()) ::
  Kashka.Http.t()

Build connection to exsisting consumer. This function mostly used in tests when there is only one kafka rest proxy instance

Parameters

conn: connection or url for rest proxy instance
group: consumer group name
name: consumer instance_id

offsets(conn, partitions)

offsets(Kashka.Http.conn(), [partition()]) ::
  {:ok, Kashka.Http.t(), [any()]} | http_error()

Requests offsets for consumer. See Confluent REST Proxy API Reference

Parameters

conn: connection or url for created consumer
partitions: a list of partitions like [%{topic: "test", partition: 0}]

positions_end(conn, partitions)

positions_end(Kashka.Http.t(), [partition()]) ::
  {:ok, Kashka.Http.t()} | http_error()

Seek to the last offset for each of the given partitions. See Confluent REST Proxy API Reference

Parameters

conn: connection or url for created consumer

produce(conn, topic, records, format \\ :json)

produce(
  Kashka.Http.conn(),
  String.t(),
  [json_records() | binary_records()],
  :json | :binary
) :: {:ok, Kashka.Http.t()} | http_error()

Produce messages to a topic. See Confluent REST Proxy API Reference

Parameters

conn: connection or url for rest proxy
topic: topic name
records: records list
format: records format json or binary

topics(conn)

topics(Kashka.Http.conn()) ::
  {:ok, Kashka.Http.t(), [String.t()]} | http_error()

Requests all topics list See Confluent REST Proxy API Reference

Parameters

conn: connection or url for rest proxy