OnrampImpl extends OnrampTemplate<byte[], byte[]> implements Onramp { @Override protected Future<Boolean> sendEncodedEvent(Event<byte[], byte[]> event, SchemaVersion schemaVersion) throws InvalidKeyException { ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(getRoad().getTopicName(), event.getKey(), event.getMessage()); Future<RecordMetadata> future = kafkaProducer.send(record, this::updateMetrics); return Futures.lazyTransform(future, metadata -> true); } OnrampImpl(OnrampMetrics metrics, Producer<byte[], byte[]> kafkaProducer, Road road); @Override SchemaVersion getSchemaVersion();  }
@SuppressWarnings("unchecked") @Test public void sendEncodedEvent_UpdateMetrics_Success() throws InvalidKeyException { RecordMetadata metadata = new RecordMetadata(null, 0, 0, 0, Long.valueOf(0), 0, 1); Exception exception = null; when(kafkaProducer.send(any(ProducerRecord.class), any(Callback.class))).thenAnswer(invocation -> { ((Callback) invocation.getArgument(1)).onCompletion(metadata, exception); return future; }); underTest.sendEncodedEvent(new Event<>(null, null), null); verify(metrics).markSuccessMetrics(ROAD_NAME, 1); }
@SuppressWarnings("unchecked") @Test public void sendEncodedEvent_UpdateMetrics_Failure() throws InvalidKeyException { RecordMetadata metadata = null; Exception exception = new Exception(); when(kafkaProducer.send(any(ProducerRecord.class), any(Callback.class))).thenAnswer(invocation -> { ((Callback) invocation.getArgument(1)).onCompletion(metadata, exception); return future; }); underTest.sendEncodedEvent(new Event<>(null, null), null); verify(metrics).markFailureMetrics(ROAD_NAME); }