KafkaMRecordSource implements MRecordSource { @Override public void close() { consumer.close(); } KafkaMRecordSource(KafkaConsumer<String, UntypedValues> consumer); @SuppressWarnings("squid:S2095") // KafkaConsumer is responsible for the closing of the Deserializers  KafkaMRecordSource(String kafkaUrl, String kafkaGroup, Collection<String> topics); @SuppressWarnings("squid:S2095") // KafkaConsumer is responsible for the closing of the Deserializers  KafkaMRecordSource(String kafkaUrl, String kafkaGroup, Collection<String> topics, boolean fromBeginning);  KafkaMRecordSource(String kafkaUrl, String kafkaGroup, Collection<String> topics,
                              Map<String, Object> additionalProperties); @Override UntypedValues get(); @Override boolean hasRemaining(); @Override void init(); @Override void close(); @Override Kind getKind();  }
@Test public void close() { try{ source.close(); } catch (Exception ex){ fail("failed to close consumer"); } }