### Question:
ProcessingKafkaConsumer implements Closeable { public Consumer<K, V> getConsumer() { return consumer; } ProcessingKafkaConsumer(ProcessingConfig config);  ProcessingKafkaConsumer(ProcessingConfig config, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer);   ProcessingKafkaConsumer(ProcessingConfig config, Consumer<K, V> consumer); Optional<ConsumerRecord<K, V>> nextRecord(final long timeout); boolean ack(ConsumerRecord<K, V> record); boolean ack(TopicPartition topicPartition, long offset); boolean fail(ConsumerRecord<K, V> record); boolean fail(TopicPartition topicPartition, long offset); synchronized void resetOffsets(Map<TopicPartition, Long> offsets); synchronized void commitOffsets(); void subscribe(Collection<String> topics); Consumer<K, V> getConsumer(); @Override void close();  }### Answer:
@Test public void constructorWithoutConsumer() throws IOException { try (ProcessingKafkaConsumer<String, String> processingKafkaConsumer = new ProcessingKafkaConsumer<>(config)) { assertThat(processingKafkaConsumer.getConsumer(), is(instanceOf(KafkaConsumer.class))); } }
@Test public void constructorWithConsumer() { assertThat(processingConsumer.getConsumer(), is(consumer)); }
@Test public void constructorWithDeserializers() throws IOException { Deserializer<String> keyDeserializer = new StringDeserializer(); Deserializer<String> valueDeserializer = new StringDeserializer(); try (ProcessingKafkaConsumer<String, String> processingKafkaConsumer = new ProcessingKafkaConsumer<>(config, keyDeserializer, valueDeserializer)) { assertThat(processingKafkaConsumer.getConsumer(), is(instanceOf(KafkaConsumer.class))); } }