@Test(expected = RuntimeException.class) public void jsonSerializationProblem() throws Exception { when(mapper.writer()).thenReturn(writer); when(writer.writeValueAsString(any())).thenThrow(JsonProcessingException.class); underTest = new KafkaPatchSetEmitter("kafka-agent-topic", kafkaProducer, mapper); underTest.emit(new PatchSet("road1", new ArrayList<>())); }
@Override public void emit(PatchSet patchSet) { try { ProducerRecord<String, String> record = new ProducerRecord<>(trafficControlTopic, patchSet.getDocumentId(), writer.writeValueAsString(patchSet)); kafkaProducer.send(record, this::updateMetrics).get(); } catch (InterruptedException | ExecutionException | JsonProcessingException e) { throw new RuntimeException(e); } }
KafkaPatchSetEmitter implements PatchSetEmitter { @Override public void emit(PatchSet patchSet) { try { ProducerRecord<String, String> record = new ProducerRecord<>(trafficControlTopic, patchSet.getDocumentId(), writer.writeValueAsString(patchSet)); kafkaProducer.send(record, this::updateMetrics).get(); } catch (InterruptedException | ExecutionException | JsonProcessingException e) { throw new RuntimeException(e); } } }
KafkaPatchSetEmitter implements PatchSetEmitter { @Override public void emit(PatchSet patchSet) { try { ProducerRecord<String, String> record = new ProducerRecord<>(trafficControlTopic, patchSet.getDocumentId(), writer.writeValueAsString(patchSet)); kafkaProducer.send(record, this::updateMetrics).get(); } catch (InterruptedException | ExecutionException | JsonProcessingException e) { throw new RuntimeException(e); } } KafkaPatchSetEmitter(String trafficControlTopic, Producer<String, String> kafkaProducer);  KafkaPatchSetEmitter(String trafficControlTopic, Producer<String, String> kafkaProducer, ObjectMapper mapper); }
KafkaPatchSetEmitter implements PatchSetEmitter { @Override public void emit(PatchSet patchSet) { try { ProducerRecord<String, String> record = new ProducerRecord<>(trafficControlTopic, patchSet.getDocumentId(), writer.writeValueAsString(patchSet)); kafkaProducer.send(record, this::updateMetrics).get(); } catch (InterruptedException | ExecutionException | JsonProcessingException e) { throw new RuntimeException(e); } } KafkaPatchSetEmitter(String trafficControlTopic, Producer<String, String> kafkaProducer);  KafkaPatchSetEmitter(String trafficControlTopic, Producer<String, String> kafkaProducer, ObjectMapper mapper); @Override void emit(PatchSet patchSet); }
KafkaPatchSetEmitter implements PatchSetEmitter { @Override public void emit(PatchSet patchSet) { try { ProducerRecord<String, String> record = new ProducerRecord<>(trafficControlTopic, patchSet.getDocumentId(), writer.writeValueAsString(patchSet)); kafkaProducer.send(record, this::updateMetrics).get(); } catch (InterruptedException | ExecutionException | JsonProcessingException e) { throw new RuntimeException(e); } } KafkaPatchSetEmitter(String trafficControlTopic, Producer<String, String> kafkaProducer);  KafkaPatchSetEmitter(String trafficControlTopic, Producer<String, String> kafkaProducer, ObjectMapper mapper); @Override void emit(PatchSet patchSet);  }