Connection leak possibility on some plugin tasks

[loicmathieu]

Expected Behavior

Using try/catch on connection instantiation eveywhere to avoid possible connection leak

Actual Behaviour

Plugins that open connection usually don't do it in a reliant manner using try/catch/finally risking connection leak in case of exception.

An example can be found in the Kafka Consume task.

We need to check all connection-based plugins for such issues

[anna-geller]

closing since this was open for ~2 years

@loicmathieu feel free to reopen with more details if this is still relevant