### Question:
KafkaSink implements SelfDocumentingSink<T> { KafkaSink(Registry registry, Func1<T, byte[]> encoder) { this.encoder = encoder; this.registry = registry; } KafkaSink(Registry registry, Func1<T, byte[]> encoder); @Override void call(Context context, PortRequest ignore, Observable<T> dataO); @Override List<ParameterDefinition<?>> getParameters(); @Override Metadata metadata();  }### Answer:
@Test public void testKafkaSink() throws InterruptedException { String testTopic = "testTopic" + topicNum.incrementAndGet(); int numPartitions = 1; kafkaServer.createTopic(testTopic, numPartitions); int numMessages = 10; KafkaSink<String> kafkaSink = new KafkaSink<>(new NoopRegistry(), s -> s.getBytes()); Context context = mock(Context.class); Parameters params = ParameterTestUtils.createParameters(KafkaSinkJobParameters.TOPIC, testTopic); when(context.getParameters()).then((Answer<Parameters>) invocation -> params); when(context.getWorkerInfo()).then((Answer<WorkerInfo>) invocation -> new WorkerInfo("testJobName", "testJobName-1", 1, 0, 1, MantisJobDurationType.Perpetual, "1.1.1.1")); when(context.getJobId()).then((Answer<String>) invocation -> "testJobName-1"); kafkaSink.call(context, mock(PortRequest.class), Observable.range(0, numMessages).map(x -> String.valueOf(x))); List<String> messages = kafkaServer.readAllMessages(testTopic); LOGGER.info("got {}", messages); assertEquals(numMessages, messages.size()); for (int i = 0; i < numMessages; i++) { assertEquals(i, Integer.parseInt(messages.get(i))); } kafkaServer.deleteTopic(testTopic); }