Lost ACKs for redelivered messages in RMQSource

As described in the FLINK-20244, the redelivered messages are not acknowledged properly (only applicable when autoAck is disabled). When used with a prefetch count in the consumer it may even lead to stop the source to consume any more messages.

The solution (proposed in FLINK-20244) should resolve the issue. All successfully consumed RMQ messages should be acknowledged, regardless of whether the message is ignored or processed further in the pipeline. The sessionIds.add(deliveryTag) (RMQSource.java#L423) should be called before checking if the message has already been processed.