How to properly restart a kafka s3 sink connect?

Question

I started a kafka s3 sink connector (bundle connector from confluent package) since 1 May. It works fine until 8 May. Checking the status, it tells that some aws exception crashes this connector. This should not be a big problem, so I want to restore it.

I tried the following steps:

1. I POST /connectors/s3sink/restart . Then I saw the connector is in RUNNING mode, but the task is still FAIL.
2. Then I PUT /connectors/s3sink/task/0/restart. Ok, now the task is in RUNNING mode.

But then I tail the log, I found it starts to rewrite the old data, such as 3 May data. And it messed the old data!

So, does connect restart REST API reset the offset? I thought it will save the offset and just start from the offset it fails.

And how to restart a failed connector task correctly? By deleting those PODs? (using kubernetes), or by REST /task/0/restart? When should I use /connectors/s3sink/restart?

Unbonnet

Answer 1

/connector/:name/restart is a rolling restart operation on the worker leader that needs to propagate to all worker server tasks in async fashion. So, you need to ensure network connection between the leader worker and all others.

/connector/:name/task/:num/restart will send request straight to that worker, restarting the thread.

Restart should not reset the offset since they are stored in the consumer offsets topic for that connect cluster. If anything, the tasks were not able to commit offsets back to the __consumer_offsets topic, but you should see logs for that.

Hopscotch

Thanks, the reason is the offset retention time, I restart it after one day – Unbonnet

Not sure what you mean. I'm currently managing tens of S3 connects with a week retention on the topics, and don't need to be restarted that often (unless we get an error) – Hopscotch

In our case, the 2nd command in your answer is throwing a 404. What could be the issue? And is restarting the task the same as restarting the connector process or is it a separate thing? If separate, please tell us how to restart a connector process too. Thank you! @Hopscotch – Parimutuel

@Cyber Are you actually passing in a number for the task value? Restarting the process would be done on the host by restarting the process (for example, with systemctl restart), which would stop all tasks on the host, not for only a specific connector – Hopscotch

@Hopscotch thank you for the response. Actually, we don't have access to the host. And the host is actually a pod on kubernetes. But we get your point. We'll do that. Thank you! – Parimutuel