anchor.diff:                               end
                        end
                end,
  -{ok,Fd} = file:open(LogFile, [read]),
  -Loop = fun TLoop() ->
  -               case file:read_line(Fd) of
  -                   {ok, Data} ->
  -                       case HandleLine(Data) of
  -                           {fail, skip} ->
  -                               skip;
  -                           {fail, Reason} ->
  -                               INFO("HandleFail:reason:~p,Data:~p~n", [Reason, Data]);
  -                           ok ->
  -                               timer:sleep(1),
  -                               ok
  -                       end,
  -                       TLoop();
  -                   Other ->
  -                       INFO("Other:~p~n", [Other])
  -               end
  -       end,
  -Loop(),
  -file:close(Fd),
  -
  -
  +case whereis(recover_kafka_from_log) of
  +    undefined ->
  +        spawn(fun() ->
  +                      register(recover_kafka_from_log, self()),
  +                      {ok,Fd} = file:open(LogFile, [read]),
  +                      Loop = fun TLoop() ->
  +                                     case file:read_line(Fd) of
  +                                         {ok, Data} ->
  +                                             case HandleLine(Data) of
  +                                                 {fail, skip} ->
  +                                                     skip;
  +                                                 {fail, Reason} ->
  +                                                     ERROR_MSG("HandleFail:reason:~p,Data:~s~n", [Reason, Data]);
  +                                                 ok ->
  +                                                     ok
  +                                             end,
  +                                             TLoop();
  +                                         Other ->
  +                                             INFO("Other:~p~n", [Other])
  +                                     end
  +                             end,
  +                      Loop(),
  +                      file:close(Fd),
  +                      INFO("RepairKafka: finished~n", [])
  +              end);
  +    _ ->
  +        io:format("error:another_repair_is_running~n")
  +end,
   ok.
  
positive.code:                         end
                  end,
  {ok, LogKafkaConfig} = application:get_env(message_store, log_kafka),
  OutgoingMsgTopic = proplists:get_value(kafka_outgoing_msg_topic, LogKafkaConfig, <<"ejabberd-chat-messages">>),
  IncomingMsgTopic = proplists:get_value(kafka_incoming_msg_topic, LogKafkaConfig, <<"ejabberd-chat-recvmsgs">>),
  OfflineMsgTopic = proplists:get_value(kafka_offline_msg_topic, LogKafkaConfig, <<"ejabberd-chat-offlines">>),
  
negative.code: 
  %% input: none
  %%
  %% op: recover ekaf
  %%
  %% e.g.: ./erl_expect -sname ejabberd@sdb-ali-hangzhou-ejabberd3 -setcookie 'LTBEXKHWOCIRRSEUNSYS' common/recover_kafka_from_log.erl "/data/apps/opt/ejabberd/var/log/ejabberd.log.0" 
  
  echo(off),
  case Args of
  	[LogFile] ->
  		IsWriteKafka = true;
  