Change the cancellation detection strategy

[boma96]

Cancellation detection was primitive(i.e. cancellation was logged only if TaskCancelledException was thrown). This had several issues.

• Some libraries throw OperationCancelledException if the task is cancelled (hence why we now use OperationCancelledException instead of TaskCancelledException). Also OperationCancelledException is base of TaskCancelledException which means we can handle the later.
• We also want to make sure that cancellation request comes from Kafka topic(instead of someone just throwing OperationCancelledException) hence why we introduce additional condition.