### Question:
ProcessingKafkaConsumer implements Closeable { public boolean ack(ConsumerRecord<K, V> record) { if (record == null) throw new IllegalArgumentException("record cannot be null"); return ack(new TopicPartition(record.topic(), record.partition()), record.offset()); } 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(expected = IllegalArgumentException.class) public void ack_nullTopicPartition() { processingConsumer.ack(null, 0); }
@Test public void ack() { long previousAckCount = ProcessingKafkaConsumer.ACK_METER.count(); Optional<ConsumerRecord<String, String>> optional = processingConsumer.nextRecord(POLL_TIME); assertThat("optional is not present", optional.isPresent(), is(true)); assertThat(processingConsumer.ack(topicPartition, offset), is(true)); assertThat(processingConsumer.getCommittableOffsets().get(topicPartition), is(new OffsetAndMetadata(offset + 1))); assertThat(ProcessingKafkaConsumer.ACK_METER.count(), is(previousAckCount + 1)); }
@Test public void ack_invalidTopicPartition() { final ConsumerRecord<String, String> invalidPartition = new ConsumerRecord<>("notARealTopic", 0,0,"key", "value"); assertThat(processingConsumer.ack(invalidPartition), is(false)); }
@Test(expected = IllegalArgumentException.class) public void ack_topicPartitionIsNull() { processingConsumer.ack(null, 0); }
@Test(expected = IllegalArgumentException.class) public void ack_topicPartitionIsNull2() { processingConsumer.ack(null); }