Hao/ts 9303

.gitignore

@@ -11,4 +11,8 @@ log
 rebar
 .rebar
 *.dump
-bin/tmp
+bin/tmp
+logs/
+rel/
+.idea/
+*.iml

Makefile

@@ -1,25 +1,13 @@
 COOKIE ?= KAFKERL-EXAMPLE
 CONFIG ?= rel/kafkerl.app.config
 ERL ?= erl
-RUN := ${ERL} -pa ebin -pa deps/*/ebin -smp enable -s lager -setcookie ${COOKIE} -config ${CONFIG} -boot start_sasl ${ERL_ARGS}
+RUN := ${ERL} -pa ebin -pa deps/*/ebin -smp enable -s lager -s kafkerl -setcookie ${COOKIE} -config ${CONFIG} -boot start_sasl ${ERL_ARGS}
 NODE ?= kafkerl
 CT_ARGS ?= "-vvv"
-CT_LOG ?= /logs/ct
-ERLARGS=-pa ${DEPS} -pa ${APPS} -smp enable -boot start_sasl -args_file ${VM_ARGS} -s lager -s redis_config
-TEST_ERL_ARGS ?= ${ERLARGS} -args_file ${TEST_VM_ARGS} -config ${TEST_CONFIG}
+ERLARGS=-config ${CONFIG}
+TEST_ERL_ARGS ?= ${ERLARGS}
 REBAR ?= "rebar"
 
-ifdef CT_SUITES
-	CT_SUITES_="suites=${CT_SUITES}"
-else
-	CT_SUITES_=""
-endif
-ifdef CT_CASE
-	CT_CASE_="case=${CT_CASE}"
-else
-	CT_CASE_=""
-endif
-
 all:
 	${REBAR} get-deps compile
 
@@ -74,8 +62,4 @@ test: tests
 
 tests:
 	@${REBAR} compile skip_deps=true
-	@rm -rf ${CT_LOG}
-	@mkdir -p ${CT_LOG}
-	@ERL_FLAGS="${TEST_ERL_ARGS}" \
-	ERL_AFLAGS="${TEST_ERL_ARGS}" \
-	${REBAR} -v 3 skip_deps=true ${CT_SUITES_} ${CT_CASE_} ct
+	${REBAR} -v 3 skip_deps=true ct

README.md

@@ -1,9 +1,9 @@
-kafkerl v1.0.2
+kafkerl v3.0.0
 ==============
 [![Gitter](https://badges.gitter.im/Join Chat.svg)](https://gitter.im/HernanRivasAcosta/kafkerl?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 
-Apache Kafka 0.8 high performance producer for erlang.
-Developed thanks to the support and sponsorship of [TigerText](http://www.tigertext.com/).
+Apache Kafka 0.8.2 high performance producer/consumer for erlang.
+Developed thanks to the support and sponsorship of [TigerText](http://www.tigertext.com/) and [Inaka](https://github.com/inaka/).
 
 ##Features (aka, why kafkerl?)
  - Fast binary creation.
@@ -12,12 +12,12 @@ Developed thanks to the support and sponsorship of [TigerText](http://www.tigert
  - Messages are not lost but cached before sending to kafka.
  - Handles server side errors and broker/leadership changes.
  - Flexible API allows consumer of messages to define pids, funs or M:F pairs as callbacks for the received messages.
+ - Simple yet flexible consumer API to retrieve the messages from Kafka.
 
 ##Missing features (aka, what I am working on but haven't finished yet)
- - Though the library can parse kafka messages, the consumers are not implemented in this version.
  - There is no communication with Zookeeper.
  - Tests suites.
 
 
 
-Special thanks to [@nitzanharel](https://github.com/nitzanharel) who found some really nasty bugs and helped me understand the subtleties of kafka's design and to the rest of the [TigerText](http://www.tigertext.com/) team for their support and code reviews.
+Special thanks to [@nitzanharel](https://github.com/nitzanharel) who found some really nasty bugs and helped me understand the subtleties of kafka's design and to the rest of the [TigerText](http://www.tigertext.com/) and [Inaka](https://github.com/inaka/) teams for their support and code reviews.

bin/create_test_topics.sh

@@ -23,6 +23,6 @@ done
 # make sure the path is defined
 if [ ! -d "${d}" ]; then echo "invalid kafka path ${d}" ; exit 1 ; fi
 
-"${d}/bin/kafka-topics.sh" --zookeeper localhost:2181 --create --topic test1 --partitions 3 --replication-factor 3
-"${d}/bin/kafka-topics.sh" --zookeeper localhost:2181 --create --topic test2 --partitions 3 --replication-factor 3
-"${d}/bin/kafka-topics.sh" --zookeeper localhost:2181 --create --topic test3 --partitions 4 --replication-factor 3
+"${d}/bin/kafka-topics.sh" --zookeeper localhost:2181 --create --topic kafkerl_test1 --partitions 1 --replication-factor 3
+"${d}/bin/kafka-topics.sh" --zookeeper localhost:2181 --create --topic kafkerl_test2 --partitions 2 --replication-factor 3
+"${d}/bin/kafka-topics.sh" --zookeeper localhost:2181 --create --topic kafkerl_test3 --partitions 3 --replication-factor 3

bin/delete_test_topics.sh

@@ -0,0 +1 @@
+rm -rf tmp

bin/stop_all_brokers.sh

@@ -1 +1 @@
-ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}' | xargs kill -15
+ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}' | xargs kill -9

bin/stop_zk.sh

@@ -1 +1 @@
-ps ax | grep -i 'zookeeper' | grep -v grep | awk '{print $1}' | xargs kill -15
+ps ax | grep -i 'zookeeper' | grep -v grep | awk '{print $1}' | xargs kill -9

include/kafkerl.hrl

@@ -1,94 +1,99 @@
-%% Constants
-% Misc
--define(ETS_BUFFER, ets_buffer).
--define(DEFAULT_TCP_OPTS, lists:sort([{mode, binary}, {packet, 0}])).
-% Compression
--define(COMPRESSION_NONE,   none).
--define(COMPRESSION_GZIP,   gzip).
--define(COMPRESSION_SNAPPY, snappy).
--define(KAFKERL_COMPRESSION_TYPES, [?COMPRESSION_NONE,
-                                    ?COMPRESSION_GZIP,
-                                    ?COMPRESSION_SNAPPY]).
-% API keys
--define(PRODUCE_KEY,  0).
--define(FETCH_KEY,    1).
--define(OFFSET_KEY,   2).
--define(METADATA_KEY, 3).
-
-%% Common
--type error_code()     :: -1..16.
--type correlation_id() :: non_neg_integer().
-
-%% Connection
--type address_host()   :: string().
--type address_port()   :: 1..65535.
--type socket_address() :: {address_host(), address_port()}.
--type broker_id()      :: integer().
--type broker()         :: {broker_id(), socket_address()}.
-
-%% Configuration
--type compression() :: ?COMPRESSION_NONE |
-                       ?COMPRESSION_GZIP |
-                       ?COMPRESSION_SNAPPY.
-
-%% Requests
--type client_id()      :: binary().
--type topic()          :: binary().
--type partition()      :: integer().
--type payload()        :: binary() | [binary()].
--type basic_message()  :: {topic(), partition(), payload()}.
--type merged_message() :: basic_message() |
-                          {topic(), [{partition(), payload()}]} |
-                          [merged_message()].
-
--type fetch_offset()    :: integer().
--type fetch_max_bytes() :: integer().
--type fetch_partition() :: {partition(), fetch_offset(), fetch_max_bytes()} |
-                           [fetch_partition()].
--type fetch_request()   :: {topic(), fetch_partition()} |
-                           [fetch_request()].
-
-%% Reponses
--type error() :: {error, atom() | {atom(), any()}}.
-
-%% Produce responses
--type produce_partition() :: {partition(), error_code(), integer()}.
--type produce_topic()     :: {topic(), [produce_partition()]}.
--type produce_response()  :: {ok, correlation_id(), [produce_topic()]}.
-
-%% Fetch responses
--type messages()       :: [{topic(), [{{partition(), integer()}, [binary()]}]}].
--type fetch_state()    :: {binary(), integer(), [any()]}.
--type fetch_response() :: {ok, integer(), messages()} |
-                          {incomplete, integer(), messages(), fetch_state()} |
-                          error().
-
-%% Metadata responses
--type leader()             :: integer().
--type replica()            :: integer().
--type isr()                :: integer().
--type partition_metadata() :: {error_code(), partition(), broker_id(),
-                               [replica()], [isr()]}.
--type topic_metadata()     :: {error_code(), topic(), [partition_metadata()]}.
--type metadata()           :: {[broker()], [topic_metadata()]}.
--type metadata_response()  :: {ok, correlation_id(), metadata()} |
-                              error().
-
 %% Error codes
--define(NO_ERROR, 0).
+-define(UNKNOWN_SERVER_ERROR, -1).
+-define(NONE, 0).
 -define(OFFSET_OUT_OF_RANGE, 1).
--define(INVALID_MESSAGE, 2).
+-define(CORRUPT_MESSAGE, 2).
 -define(UNKNOWN_TOPIC_OR_PARTITION, 3).
--define(INVALID_MESSAGE_SIZE, 4).
+-define(INVALID_FETCH_SIZE, 4).
 -define(LEADER_NOT_AVAILABLE, 5).
--define(NOT_LEADER_FOR_PARTITION, 6).
--define(REQUEST_TIMEDOUT, 7).
+-define(NOT_LEADER_OR_FOLLOWER, 6).
+-define(REQUEST_TIMED_OUT, 7).
 -define(BROKER_NOT_AVAILABLE, 8).
 -define(REPLICA_NOT_AVAILABLE, 9).
--define(MESSAGE_SIZE_TOO_LARGE, 10).
+-define(MESSAGE_TOO_LARGE, 10).
 -define(STALE_CONTROLLER_EPOCH, 11).
 -define(OFFSET_METADATA_TOO_LARGE, 12).
--define(OFFSETS_LOAD_IN_PROGRESS_CODE, 14).
--define(CONSUMER_COORDINATOR_NOT_AVAILABLE_CODE, 15).
--define(NOT_COORDINATOR_FOR_CONSUMER_CODE, 16).
--define(UNKNOWN, -1).
+-define(NETWORK_EXCEPTION, 13).
+-define(COORDINATOR_LOAD_IN_PROGRESS, 14).
+-define(COORDINATOR_NOT_AVAILABLE, 15).
+-define(NOT_COORDINATOR, 16).
+-define(INVALID_TOPIC_EXCEPTION, 17).
+-define(RECORD_LIST_TOO_LARGE, 18).
+-define(NOT_ENOUGH_REPLICAS, 19).
+-define(NOT_ENOUGH_REPLICAS_AFTER_APPEND, 20).
+-define(INVALID_REQUIRED_ACKS, 21).
+-define(ILLEGAL_GENERATION, 22).
+-define(INCONSISTENT_GROUP_PROTOCOL, 23).
+-define(INVALID_GROUP_ID, 24).
+-define(UNKNOWN_MEMBER_ID, 25).
+-define(INVALID_SESSION_TIMEOUT, 26).
+-define(REBALANCE_IN_PROGRESS, 27).
+-define(INVALID_COMMIT_OFFSET_SIZE, 28).
+-define(TOPIC_AUTHORIZATION_FAILED, 29).
+-define(GROUP_AUTHORIZATION_FAILED, 30).
+-define(CLUSTER_AUTHORIZATION_FAILED, 31).
+-define(INVALID_TIMESTAMP, 32).
+-define(UNSUPPORTED_SASL_MECHANISM, 33).
+-define(ILLEGAL_SASL_STATE, 34).
+-define(UNSUPPORTED_VERSION, 35).
+-define(TOPIC_ALREADY_EXISTS, 36).
+-define(INVALID_PARTITIONS, 37).
+-define(INVALID_REPLICATION_FACTOR, 38).
+-define(INVALID_REPLICA_ASSIGNMENT, 39).
+-define(INVALID_CONFIG, 40).
+-define(NOT_CONTROLLER, 41).
+-define(INVALID_REQUEST, 42).
+-define(UNSUPPORTED_FOR_MESSAGE_FORMAT, 43).
+-define(POLICY_VIOLATION, 44).
+-define(OUT_OF_ORDER_SEQUENCE_NUMBER, 45).
+-define(DUPLICATE_SEQUENCE_NUMBER, 46).
+-define(INVALID_PRODUCER_EPOCH, 47).
+-define(INVALID_TXN_STATE, 48).
+-define(INVALID_PRODUCER_ID_MAPPING, 49).
+-define(INVALID_TRANSACTION_TIMEOUT, 50).
+-define(CONCURRENT_TRANSACTIONS, 51).
+-define(TRANSACTION_COORDINATOR_FENCED, 52).
+-define(TRANSACTIONAL_ID_AUTHORIZATION_FAILED, 53).
+-define(SECURITY_DISABLED, 54).
+-define(OPERATION_NOT_ATTEMPTED, 55).
+-define(KAFKA_STORAGE_ERROR, 56).
+-define(LOG_DIR_NOT_FOUND, 57).
+-define(SASL_AUTHENTICATION_FAILED, 58).
+-define(UNKNOWN_PRODUCER_ID, 59).
+-define(REASSIGNMENT_IN_PROGRESS, 60).
+-define(DELEGATION_TOKEN_AUTH_DISABLED, 61).
+-define(DELEGATION_TOKEN_NOT_FOUND, 62).
+-define(DELEGATION_TOKEN_OWNER_MISMATCH, 63).
+-define(DELEGATION_TOKEN_REQUEST_NOT_ALLOWED, 64).
+-define(DELEGATION_TOKEN_AUTHORIZATION_FAILED, 65).
+-define(DELEGATION_TOKEN_EXPIRED, 66).
+-define(INVALID_PRINCIPAL_TYPE, 67).
+-define(NON_EMPTY_GROUP, 68).
+-define(GROUP_ID_NOT_FOUND, 69).
+-define(FETCH_SESSION_ID_NOT_FOUND, 70).
+-define(INVALID_FETCH_SESSION_EPOCH, 71).
+-define(LISTENER_NOT_FOUND, 72).
+-define(TOPIC_DELETION_DISABLED, 73).
+-define(FENCED_LEADER_EPOCH, 74).
+-define(UNKNOWN_LEADER_EPOCH, 75).
+-define(UNSUPPORTED_COMPRESSION_TYPE, 76).
+-define(STALE_BROKER_EPOCH, 77).
+-define(OFFSET_NOT_AVAILABLE, 78).
+-define(MEMBER_ID_REQUIRED, 79).
+-define(PREFERRED_LEADER_NOT_AVAILABLE, 80).
+-define(GROUP_MAX_SIZE_REACHED, 81).
+-define(FENCED_INSTANCE_ID, 82).
+-define(ELIGIBLE_LEADERS_NOT_AVAILABLE, 83).
+-define(ELECTION_NOT_NEEDED, 84).
+-define(NO_REASSIGNMENT_IN_PROGRESS, 85).
+-define(GROUP_SUBSCRIBED_TO_TOPIC, 86).
+-define(INVALID_RECORD, 87).
+-define(UNSTABLE_OFFSET_COMMIT, 88).
+-define(THROTTLING_QUOTA_EXCEEDED, 89).
+-define(PRODUCER_FENCED, 90).
+-define(RESOURCE_NOT_FOUND, 91).
+-define(DUPLICATE_RESOURCE, 92).
+-define(UNACCEPTABLE_CREDENTIAL, 93).
+-define(INCONSISTENT_VOTER_SET, 94).
+-define(INVALID_UPDATE_VERSION, 95).
+-define(FEATURE_UPDATE_FAILED, 96).

rebar.config

@@ -16,8 +16,8 @@
             warn_untyped_record,
             {i, "include"}]}.
 {deps,
- [{parse_trans, ".*", {git, "[email protected]:uwiger/parse_trans.git", "master"}},
-  {lager,       ".*", {git, "[email protected]:basho/lager.git", "master"}},
-  {epocxy,   "0.9.7", {git, "[email protected]:duomark/dk_cxy.git", {tag, "0.9.7"}}},
-  {validerl,    ".*", {git, "https://github.com/HernanRivasAcosta/validerl.git",
-                       "master"}}]}.
+ [{parse_trans, ".*",   {git, "[email protected]:tigertext/parse_trans.git", "master"}},
+  {lager,       ".*",   {git, "[email protected]:tigertext/lager.git",    {tag, "2.1.1"}}},
+  {epocxy,      ".*",   {git, "[email protected]:tigertext/epocxy.git",   "master"}},
+  {eper,        "0.69", {git, "git@github.com:tigertext/eper.git",     "HEAD"}},
+  {validerl,    ".*",   {git, "[email protected]:tigertext/validerl.git", "master"}}]}.

rel/kafkerl.app.config

@@ -1,19 +1,22 @@
 [{lager, [{colored, true},
           {handlers, [{lager_console_backend, [debug,true]}]}]},
- {kafkerl, [%{gen_server_name, kafkerl_client},
-            {disabled, false},
-            {conn_config, [{brokers, [{"localhost", 9090},
-                                      {"localhost", 9091},
-                                      {"localhost", 9092}]},
+ {kafkerl, [{disabled, false},
+            {conn_config, [{brokers, [{"localhost", 9092}, {"localhost", 9094}
+                                      ]},
                            {client_id, kafkerl_client}, % Sent to kafka
                            {max_broker_retries, 2},
                            {broker_tcp_timeout, 1000},
-                           {max_metadata_retries, -1},
+                           {max_metadata_retries, 3},
                            {assume_autocreate_topics, true},
-                           {metadata_tcp_timeout, 1000},
+                           {metadata_tcp_timeout, 5000},
                            {max_queue_size, 20}, % In items, per topic/partition
                            {max_time_queued, 5}, % In seconds
-                           {metadata_request_cooldown, 1500} % In milliseconds
-                           ]},
-            {topics, [test1, test2, test3]},
-            {tests, [{kafka_installation, "~/kafka"}]}]}].
+                           {metadata_request_cooldown, 3500}, % In milliseconds
+                           {consumer_min_bytes, 1},
+                           {consumer_max_wait, 1500},
+                           {dump_location, "dumps/"},
+                           {flush_to_disk_every, 20000}, % In milliseconds
+                           {max_buffer_size, 5000},
+                           {save_messages_for_bad_topics, true}]},
+            {topics, []},
+            {tests, [{kafkerl_path, "/usr/local/Cellar/kafka/0.8.2.2/"}]}]}].

src/kafkerl.app.src

@@ -1,8 +1,13 @@
 {application, kafkerl,
  [{description, []},
-  {vsn, "1"},
+  {vsn, "1.5"},
   {registered, []},
-  {applications, [kernel, stdlib]},
+  {applications,
+   [
+    kernel,
+    stdlib,
+    validerl
+   ]},
   {mod, {kafkerl, []}},
   {env, []},
-  {lager, [{handlers, [{lager_console_backend, info}]}]}]}.
+  {lager, [{handlers, [{lager_console_backend, info}]}]}]}.