### Question:
MockKafka implements Kafka<K, V> { @Override public FallibleMockConsumer<K, V> getConsumer(Properties overrides) { return getConsumer(new Properties(), overrides); } MockKafka();  MockKafka(int maxPartitions, int maxHistory); MockKafka<K, V> withSendCallbackExceptionGenerator(ExceptionGenerator<ProducerRecord<K, V>, Exception> sendCallbackExceptionGenerator); MockKafka<K, V> withSendRuntimeExceptionGenerator(ExceptionGenerator<ProducerRecord<K, V>, RuntimeException> sendRuntimeExceptionGenerator); MockKafka<K, V> withCommitExceptionGenerator(ExceptionGenerator<Map<TopicPartition, OffsetAndMetadata>, Exception> commitExceptionGenerator); MockKafka<K, V> withAdminClientFactory(Supplier<AdminClient> adminClientFactory); MockKafka<K, V> withRecordMapper(BiFunction<ProducerRecord<K, V>, RecordMetadata, ConsumerRecord<K, V>> recordMapper); @Override void describeProducer(LogLine logLine, Properties defaults, Properties overrides); @Override FallibleMockProducer<K, V> getProducer(Properties overrides); @Override FallibleMockProducer<K, V> getProducer(Properties defaults, Properties overrides); List<ConsumerRecord<K, V>> getBacklog(); @Override void describeConsumer(LogLine logLine, Properties defaults, Properties overrides); @Override FallibleMockConsumer<K, V> getConsumer(Properties overrides); @Override FallibleMockConsumer<K, V> getConsumer(Properties defaults, Properties overrides); @Override AdminClient getAdminClient(); @Override String toString();  }### Answer:
@Test public void testConsumeInterrupted() { final MockKafka<String, String> kafka = new MockKafka<>(2, 1); final SerdeProps props = new SerdeProps(SerdePair.STRING); final Consumer<String, String> consumer = kafka.getConsumer(props.consumer()); consumer.subscribe(Arrays.asList("different")); Thread.currentThread().interrupt(); try { assertEquals(0, consumer.poll(Duration.ofMillis(10_000)).count()); } finally { assertTrue(Thread.interrupted()); } }
@Test public void testPartitionsFor_nonExistentTopic() { final MockKafka<String, String> kafka = new MockKafka<>(1, 1); final SerdeProps props = new SerdeProps(SerdePair.STRING); final Consumer<String, String> consumer = kafka.getConsumer(props.consumer()); final List<PartitionInfo> partitions = consumer.partitionsFor("topic"); assertEquals(1, partitions.size()); }
@Test public void testCommitAsync() { final MockKafka<String, String> kafka = new MockKafka<>(1, 1); final SerdeProps props = new SerdeProps(SerdePair.STRING); final Consumer<String, String> consumer = kafka.getConsumer(props.consumer()); final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(); final OffsetCommitCallback callback = mock(OffsetCommitCallback.class); consumer.commitAsync(offsets, callback); wait.until(() -> verify(callback).onComplete(notNull(), isNull())); }