@Test public void sendPatchSet() { when(kafkaProducer.send(ArgumentMatchers.<ProducerRecord<String, String>> any(), ArgumentMatchers.any())) .thenReturn(resultFuture); 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);  }