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()); }  ErrorHandler getErrorHandler(); void handleExpectedError(IncomingMessageEnvelope envelope, Exception e); @Override void init(Config config, TaskContext context); @Override final void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator); @Override final void close(); static void run(String jobName);  }
@Test public void getMessageIdFromSource() { MyTask myTask = new MyTask(); String offset = "12345"; SystemStreamPartition ssp = new SystemStreamPartition("fakeSystem", "fakeStream", new Partition(56789)); IncomingMessageEnvelope envelope = new IncomingMessageEnvelope(ssp, offset, null, null); assertEquals("fakeStream-56789-12345", myTask.getMessageIdFromSource(envelope)); }