class BaseTask implements InitableTask, StreamTask, ClosableTask {

    protected String getMessageIdFromSource(IncomingMessageEnvelope envelope) {
        SystemStreamPartition ssp = envelope.getSystemStreamPartition();
        return String.format("%s-%d-%s", ssp.getStream(), ssp.getPartition().getPartitionId(), envelope.getOffset());
    }

}