From 87b35aeccea690001d289c67bd42099657639cd1 Mon Sep 17 00:00:00 2001
From: silviu caragea <silviu.cpp@gmail.com>
Date: Thu, 19 Jan 2023 00:07:59 +0200
Subject: [PATCH] - Migrate from lager to logger - Fix undefined functions

---
 CHANGELOG.md               |  5 +++++
 include/erlkaf_private.hrl | 19 +++----------------
 rebar.config               |  6 ++----
 rebar.lock                 | 12 +++---------
 src/erlkaf.app.src         |  3 +--
 src/erlkaf.erl             | 10 +++++-----
 6 files changed, 19 insertions(+), 36 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3325330..b9622e2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,10 @@
 ### Changelog:
 
+#### v2.1.3
+
+- Migrate from lager to erlang logger. 
+- Important bug fixing causing crashes in rare situations because of undefined functions.
+
 #### v2.1.2
 
 - Proper fix for building with mix (https://github.com/fogfish/esq/issues/20#issuecomment-1369142644)
diff --git a/include/erlkaf_private.hrl b/include/erlkaf_private.hrl
index 28b4046..f74a769 100644
--- a/include/erlkaf_private.hrl
+++ b/include/erlkaf_private.hrl
@@ -1,22 +1,9 @@
 % logs
 
--define(PRINT_MSG(Format, Args),
-    io:format(Format, Args)).
-
--define(LOG_DEBUG(Format, Args),
-    lager:debug(Format, Args)).
-
--define(LOG_INFO(Format, Args),
-    lager:info(Format, Args)).
+-include_lib("kernel/include/logger.hrl").
 
--define(LOG_WARNING(Format, Args),
-    lager:warning(Format, Args)).
-
--define(LOG_ERROR(Format, Args),
-    lager:error(Format, Args)).
-
--define(LOG_CRITICAL(Format, Args),
-    lager:critical(Format, Args)).
+-define(LOG_PRINT(Format, Args),
+    io:format(Format, Args)).
 
 % erlkaf errors
 
diff --git a/rebar.config b/rebar.config
index 324b895..8f9b22b 100644
--- a/rebar.config
+++ b/rebar.config
@@ -5,8 +5,7 @@
 {project_plugins, [rebar3_hex]}.
 
 {deps, [
-    {lager, "3.9.2"},
-    {jsone, "1.5.2"},
+    {jsone, "1.7.0"},
     {esq, "2.0.6"}
 ]}.
 
@@ -20,6 +19,5 @@
 
 {erl_opts, [
     warnings_as_errors,
-    warn_export_all,
-    {parse_transform, lager_transform}
+    warn_export_all
 ]}.
diff --git a/rebar.lock b/rebar.lock
index 6fbb487..0dae845 100644
--- a/rebar.lock
+++ b/rebar.lock
@@ -1,26 +1,20 @@
 {"1.2.0",
 [{<<"datum">>,{pkg,<<"datum">>,<<"4.6.1">>},1},
  {<<"esq">>,{pkg,<<"esq">>,<<"2.0.6">>},0},
- {<<"goldrush">>,{pkg,<<"goldrush">>,<<"0.1.9">>},1},
- {<<"jsone">>,{pkg,<<"jsone">>,<<"1.5.2">>},0},
- {<<"lager">>,{pkg,<<"lager">>,<<"3.9.2">>},0},
+ {<<"jsone">>,{pkg,<<"jsone">>,<<"1.7.0">>},0},
  {<<"pipe">>,{pkg,<<"pipes">>,<<"2.0.1">>},1},
  {<<"uid">>,{pkg,<<"uid">>,<<"1.3.4">>},1}]}.
 [
 {pkg_hash,[
  {<<"datum">>, <<"93B131203A60CFEA9FFFF6435A50DC24239F689DFEBB76E6AECF6CE689EFE8F4">>},
  {<<"esq">>, <<"9917E1A731C609B42624A4BB8594A25D537EA30E7B55D46CD46FA1B95E6DB675">>},
- {<<"goldrush">>, <<"F06E5D5F1277DA5C413E84D5A2924174182FB108DABB39D5EC548B27424CD106">>},
- {<<"jsone">>, <<"87ADEA283C9CF24767B4DEED44602989A5331156DF5D60A2660E9C9114D54046">>},
- {<<"lager">>, <<"4CAB289120EB24964E3886BD22323CB5FEFE4510C076992A23AD18CF85413D8C">>},
+ {<<"jsone">>, <<"1E3BD7D5DD44BB2EB0797DDDEA1CBF2DDAB8D9F29E499A467CA171C23F5984EA">>},
  {<<"pipe">>, <<"A2B56796C63690ED0E78BB77BB389AF250BD70AFA15A6869369DBDC11087D68F">>},
  {<<"uid">>, <<"42E30E22908E8E2FAA6227E9C261F1954CB540BE3C5A139E112369AE6CC451FC">>}]},
 {pkg_hash_ext,[
  {<<"datum">>, <<"E14340F8280FEDB1731D5CD6E9F5AEAA14B880C51F0B3DC16C42C6671C167E4D">>},
  {<<"esq">>, <<"3B798DA50C508FE93248DBBD64D3D2CB618CAB5387E66515AB83CADF2B1ABAC1">>},
- {<<"goldrush">>, <<"99CB4128CFFCB3227581E5D4D803D5413FA643F4EB96523F77D9E6937D994CEB">>},
- {<<"jsone">>, <<"170C171CE7F6DD70C858065154A3305B8564833C6DCCA17E10B676CA31EA976F">>},
- {<<"lager">>, <<"7F904D9E87A8CB7E66156ED31768D1C8E26EBA1D54F4BC85B1AA4AC1F6340C28">>},
+ {<<"jsone">>, <<"A3A33712EE6BC8BE10CFA21C7C425A299DE4C5A8533F9F931E577A6D0E8F5DBD">>},
  {<<"pipe">>, <<"623357A158E4C33EE589D4C735DDBAB9C77A04E85159192E4D42F1DC97C60BD9">>},
  {<<"uid">>, <<"F8388EF93B16A5D5F9977E1FE814AE0ACF5529B1E0EE5D7B18D23CB4C0F87EAA">>}]}
 ].
diff --git a/src/erlkaf.app.src b/src/erlkaf.app.src
index f824c08..df48140 100644
--- a/src/erlkaf.app.src
+++ b/src/erlkaf.app.src
@@ -2,12 +2,11 @@
     {description, "erlkaf - Erlang Kafka library based on librdkafka"},
     {licenses, ["MIT"]},
     {links,[{"Github","https://github.com/silviucpp/erlkaf"}]},
-    {vsn, "2.1.2"},
+    {vsn, "2.1.3"},
     {registered, []},
     {applications, [
         kernel,
         stdlib,
-        lager,
         jsone,
         esq
     ]},
diff --git a/src/erlkaf.erl b/src/erlkaf.erl
index a6c3f51..e78ff12 100644
--- a/src/erlkaf.erl
+++ b/src/erlkaf.erl
@@ -150,14 +150,14 @@ produce(ClientId, TopicName, Partition, Key, Value, Headers0, Timestamp) ->
                 ok ->
                     ok;
                 {error, ?RD_KAFKA_RESP_ERR_QUEUE_FULL} ->
-                    case erlkaf_producer:queue_event(ClientPid, TopicName, Partition, Key, Value, Headers) of
+                    case erlkaf_producer:queue_event(ClientPid, TopicName, Partition, Key, Value, Headers, Timestamp) of
                         ok ->
                             ok;
                         drop_records ->
                             ?LOG_WARNING("message: ~p dropped", [{TopicName, Partition, Key, Value, Headers}]),
                             ok;
                         block_calling_process ->
-                            produce_blocking(ClientRef, TopicName, Partition, Key, Value, Headers);
+                            produce_blocking(ClientRef, TopicName, Partition, Key, Value, Headers, Timestamp);
                         Error ->
                             Error
                     end;
@@ -177,13 +177,13 @@ get_readable_error(Error) ->
 
 %internals
 
-produce_blocking(ClientRef, TopicName, Partition, Key, Value, Headers) ->
-    case erlkaf_nif:produce(ClientRef, TopicName, Partition, Key, Value, Headers) of
+produce_blocking(ClientRef, TopicName, Partition, Key, Value, Headers, Timestamp) ->
+    case erlkaf_nif:produce(ClientRef, TopicName, Partition, Key, Value, Headers, Timestamp) of
         ok ->
             ok;
         {error, ?RD_KAFKA_RESP_ERR_QUEUE_FULL} ->
             timer:sleep(100),
-            produce_blocking(ClientRef, TopicName, Partition, Key, Value, Headers);
+            produce_blocking(ClientRef, TopicName, Partition, Key, Value, Headers, Timestamp);
         Error ->
             Error
     end.