### Question:
ProcessingKafkaConsumer implements Closeable { public boolean fail(ConsumerRecord<K, V> record) { if (record == null) throw new IllegalArgumentException("record cannot be null"); return fail(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 public void constructorWithDeserializers_nullConfig() { Deserializer<String> keyDeserializer = new StringDeserializer(); Deserializer<String> valueDeserializer = new StringDeserializer(); try { new ProcessingKafkaConsumer(null, keyDeserializer, valueDeserializer); Assert.fail("Expected IllegalArgumentException to be thrown"); } catch (IllegalArgumentException e) { } }
@Test(expected = IllegalArgumentException.class) public void fail_topicPartitionIsNull() { processingConsumer.fail(null, 0); }
@Test(expected = IllegalArgumentException.class) public void fail_topicPartitionIsNull2() { processingConsumer.fail(null); }
@Test public void fail() { long previousFailCount = ProcessingKafkaConsumer.FAIL_METER.count(); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.ack(topicPartition, record1.offset()); processingConsumer.ack(topicPartition, record3.offset()); processingConsumer.ack(new TopicPartition(record5.topic(), record5.partition()), record5.offset()); assertThat(processingConsumer.fail(topicPartition, record2.offset()), is(true)); assertThat(ProcessingKafkaConsumer.FAIL_METER.count(), is(previousFailCount + 1)); assertThat(processingConsumer.getCommittableOffsets().get(topicPartition), is(new OffsetAndMetadata(record1.offset() + 1))); assertThat(processingConsumer.getCommittableOffsets().get(new TopicPartition(record4.topic(), record4.partition())), is(nullValue())); assertThat(processingConsumer.getCommittableOffsets().get(new TopicPartition(record5.topic(), record5.partition())), is(new OffsetAndMetadata(record5.offset() + 1))); assertThat(processingConsumer.getCommittableOffsets().get(new TopicPartition(record6.topic(), record6.partition())), is(nullValue())); verify(processingConsumer.consumer).seek(topicPartition, offset); verify(consumer, never()).commitSync(anyMap()); }
@Test public void fail_topicPartitionIsNotPending() { assertThat(processingConsumer.fail(topicPartition, offset), is(false)); verify(consumer, never()).seek(topicPartition, offset); }
@Test public void fail_invalidTopicPartion() { final ConsumerRecord<String, String> invalidPartition = new ConsumerRecord<>("notARealTopic", 0, 0, "key", "value"); assertThat(processingConsumer.fail(invalidPartition), is(false)); }