Merge branch 'master' of AdRoll/erlmld

include/erlmld.hrl

@@ -25,10 +25,10 @@
 
 -record(stream_record, {
           partition_key :: binary(),
-          timestamp :: non_neg_integer(),  % approximate arrival time (ms)
+          timestamp :: undefined | non_neg_integer(),  % approximate arrival time (ms)
           delay :: non_neg_integer(),      % approximate delay between this record and tip of stream (ms)
           sequence_number :: sequence_number(),
-          data :: binary()
+          data :: term()
          }).
 
 -type worker_state() :: term().

rebar.config

@@ -42,3 +42,18 @@
                      ]}
    ]}
  ]}.
+
+{dialyzer, [
+    {warnings, [unknown, no_return, error_handling]},
+    {plt_apps, top_level_deps},
+    {plt_extra_apps, []},
+    {plt_location, local},
+    {base_plt_apps, [erts, stdlib, kernel, sasl, runtime_tools, tools]},
+    {base_plt_location, global}
+]}.
+
+{xref_checks,[
+    undefined_function_calls,
+    locals_not_used,
+    deprecated_function_calls
+]}.

src/erlmld.app.src

@@ -17,6 +17,7 @@
   {applications,
    [kernel,
     stdlib,
+    crypto,
     erlexec,
     jiffy,
     b64fast]},

src/erlmld_batch_processor.erl

@@ -33,6 +33,7 @@
 -export([initialize/3,
          ready/1,
          process_record/2,
+         checkpointed/3,
          shutdown/2]).
 
 -include("erlmld.hrl").
@@ -47,6 +48,10 @@
     %% Optional callback to call each time process_record returns a checkpoint.
     on_checkpoint :: fun((term(), shard_id()) -> term()),
 
+    %% Optional, false by default. Tells whether to log or not every successful
+    %% checkpoint from the KCL worker.
+    log_checkpoints :: boolean(),
+
     description :: term(),
     shard_id :: shard_id(),
     count = 0, % non-ignored records seen
@@ -71,12 +76,14 @@
 
 initialize(Opts, ShardId, ISN) ->
     Defaults = #{on_checkpoint => fun(_, _) -> ok end,
+                 log_checkpoints => false,
                  description => undefined,
                  enable_subsequence_checkpoints => false},
     #{description := Description,
       flusher_mod := FlusherMod,
       flusher_mod_data := FlusherModData,
       on_checkpoint := OnCheckpoint,
+      log_checkpoints := LogCheckpoints,
       flush_interval_ms := FlushIntervalMs,
       checkpoint_interval_ms := CheckpointIntervalMs,
       watchdog_timeout_ms := WatchdogTimeoutMs,
@@ -85,6 +92,7 @@ initialize(Opts, ShardId, ISN) ->
         flusher_mod = FlusherMod,
         flusher_state = FlusherMod:init(ShardId, FlusherModData),
         on_checkpoint = OnCheckpoint,
+        log_checkpoints = LogCheckpoints,
         description = Description,
         shard_id = ShardId,
         flush_interval_ms = FlushIntervalMs,
@@ -114,6 +122,14 @@ process_record(#state{last_flush_time = LastFlush,
     end,
     maybe_checkpoint(update_watchdog(NState)).
 
+checkpointed(#state{log_checkpoints = LogCheckpoints} = State,
+           SequenceNumber,
+           Checkpoint) ->
+    case LogCheckpoints of
+        true -> error_logger:info_msg("~p checkpointed at ~p (~p)~n", [State, Checkpoint, SequenceNumber]);
+        false -> ok
+    end,
+    {ok, State}.
 
 shutdown(#state{description = Description,
                 shard_id = ShardId,

src/erlmld_noisy_wrk.erl

@@ -14,6 +14,7 @@
 -export([initialize/3,
          ready/1,
          process_record/2,
+         checkpointed/3,
          shutdown/2]).
 
 -include("erlmld.hrl").
@@ -42,6 +43,12 @@ process_record(#state{shard_id = ShardId, count = Count} = State,
             {ok, State#state{count = Count + 1}}
     end.
 
+checkpointed(#state{shard_id = ShardId, count = Count} = State,
+           SequenceNumber,
+           Checkpoint
+) ->
+    io:format("~p (~p) checkpointed at ~p (~p)~n", [ShardId, Count, Checkpoint, SequenceNumber]),
+    {ok, State}.
 
 shutdown(#state{shard_id = ShardId, count = Count}, Reason) ->
     io:format("~p (~p) shutting down, reason: ~p~n", [ShardId, Count, Reason]),

src/erlmld_worker.erl

@@ -63,6 +63,10 @@
         | {ok, worker_state(), checkpoint()}
         | {error, term()}.
 
+-callback checkpointed(worker_state(), sequence_number(), checkpoint()) ->
+    {ok, worker_state()}
+        | {error, term()}.
+
 -callback shutdown(worker_state(), shutdown_reason()) ->
     ok
         | {ok, checkpoint()}

src/erlmld_wrk_statem.erl

@@ -119,23 +119,23 @@
           handler_data :: term(),
 
           %% connected socket owned by this process:
-          socket :: gen_tcp:socket(),
+          socket :: undefined | gen_tcp:socket(),
 
           %% input buffer;  responses are small and we need no output buffer:
-          buf = [] :: list(binary()),
+          buf = [] :: [binary()],
 
           %% worker state returned from handler module init:
-          worker_state :: term(),
+          worker_state :: undefined | term(),
 
           %% if true, the MLD made a processRecords call with the V2 format (supplied
           %% millisBehindLatest), so we will checkpoint using the V2 checkpoint format:
           is_v2 = false :: boolean(),
 
           %% most recent action name from the peer:
-          last_request :: binary(),
+          last_request :: undefined | binary(),
 
           %% last attempted checkpoint:
-          last_checkpoint :: checkpoint()
+          last_checkpoint :: undefined | checkpoint()
          }).
 
 -define(INTERNAL, internal).
@@ -410,22 +410,26 @@ handle_event(?INTERNAL, #{<<"action">> := <<"checkpoint">>,
 
 handle_event(?INTERNAL, #{<<"action">> := <<"checkpoint">>} = R,
              {?DISPATCH, CheckpointState},
-             #data{worker_state = {ok, WorkerState},
+             #data{handler_module = Mod,
+                   worker_state = {ok, WorkerState},
                    last_checkpoint = Checkpoint,
                    last_request = LastAction} = Data)
   when CheckpointState == ?CHECKPOINT;
        CheckpointState == ?SHUTDOWN_CHECKPOINT ->
-    %% successful checkpoint. fixme; provide indication of success to worker?
     SN = sequence_number(R),
-    error_logger:info_msg("~p checkpointed at ~p (~p)~n", [WorkerState, Checkpoint, SN]),
-    case CheckpointState of
-        ?CHECKPOINT ->
-            {next_state, ?PROCESS_RECORDS, Data};
-        ?SHUTDOWN_CHECKPOINT ->
-            success(LastAction, Data, ?SHUTDOWN)
+    case Mod:checkpointed(WorkerState, SN, Checkpoint) of
+        {ok, NWorkerState} ->
+            NData = worker_state(Data, NWorkerState),
+            case CheckpointState of
+                ?CHECKPOINT ->
+                    {next_state, ?PROCESS_RECORDS, NData};
+                ?SHUTDOWN_CHECKPOINT ->
+                    success(LastAction, NData, ?SHUTDOWN)
+          end;
+        {error, _} = Error ->
+            {stop, Error}
     end;
 
-
 %% we were processing records and we attempted to checkpoint, but failed because another
 %% worker stole our lease.  abort record processing, return a 'success' response for
 %% the current command, and read the next request, which should be a shutdown command.
@@ -586,7 +590,7 @@ next_line(Bin, #data{buf = Buf} = Data) ->
 %% and remaining data.  an "action" is a line which should have been a json-encoded map
 %% containing an "action" key.  if decoding fails with a thrown error, that error is
 %% returned as the decoded value.
--spec next_action(binary(), #data{}) -> {map() | undefined, #data{}, binary()}.
+-spec next_action(binary(), #data{}) -> {map() | undefined | {error, term()}, #data{}, binary()}.
 next_action(Bin, Data) ->
     case next_line(Bin, Data) of
         {undefined, NData, Rest} ->