KafkaPublishTask extends WorkflowSystemTask { @VisibleForTesting Object getKey(Input input) { String keySerializer = input.getKeySerializer(); if (LongSerializer.class.getCanonicalName().equals(keySerializer)) { return Long.parseLong(String.valueOf(input.getKey())); } else if (IntegerSerializer.class.getCanonicalName().equals(keySerializer)) { return Integer.parseInt(String.valueOf(input.getKey())); } else { return String.valueOf(input.getKey()); } } @Inject  KafkaPublishTask(Configuration config, KafkaProducerManager clientManager, ObjectMapper objectMapper); @Override void start(Workflow workflow, Task task, WorkflowExecutor executor); @Override boolean execute(Workflow workflow, Task task, WorkflowExecutor executor); @Override void cancel(Workflow workflow, Task task, WorkflowExecutor executor); @Override boolean isAsync();  }
@Test public void integerSerializer_integerObject() { KafkaPublishTask kPublishTask = new KafkaPublishTask(new SystemPropertiesConfiguration(), new KafkaProducerManager(new SystemPropertiesConfiguration()), objectMapper); KafkaPublishTask.Input input = new KafkaPublishTask.Input(); input.setKeySerializer(IntegerSerializer.class.getCanonicalName()); input.setKey(String.valueOf(Integer.MAX_VALUE)); Assert.assertEquals(kPublishTask.getKey(input), new Integer(Integer.MAX_VALUE)); }
@Test public void longSerializer_longObject() { KafkaPublishTask kPublishTask = new KafkaPublishTask(new SystemPropertiesConfiguration(), new KafkaProducerManager(new SystemPropertiesConfiguration()), objectMapper); KafkaPublishTask.Input input = new KafkaPublishTask.Input(); input.setKeySerializer(LongSerializer.class.getCanonicalName()); input.setKey(String.valueOf(Long.MAX_VALUE)); Assert.assertEquals(kPublishTask.getKey(input), new Long(Long.MAX_VALUE)); }
@Test public void noSerializer_StringObject() { KafkaPublishTask kPublishTask = new KafkaPublishTask(new SystemPropertiesConfiguration(), new KafkaProducerManager(new SystemPropertiesConfiguration()), objectMapper); KafkaPublishTask.Input input = new KafkaPublishTask.Input(); input.setKey("testStringKey"); Assert.assertEquals(kPublishTask.getKey(input), "testStringKey"); }