### Question:
MockKafka implements Kafka<K, V> { @Override public FallibleMockProducer<K, V> getProducer(Properties overrides) { return getProducer(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 testSingletonProducer() { final MockKafka<String, String> kafka = new MockKafka<>(1, 1); final SerdeProps props = new SerdeProps(SerdePair.STRING); final Producer<String, String> p0 = kafka.getProducer(props.producer()); assertNotNull(p0); final Producer<String, String> p1 = kafka.getProducer(props.producer()); assertSame(p0, p1); }
@Test(expected=InvalidPartitionException.class) public void testSend_invalidPartition() { final MockKafka<String, String> kafka = new MockKafka<>(1, 1); final SerdeProps props = new SerdeProps(SerdePair.STRING); final Producer<String, String> producer = kafka.getProducer(props.producer()); producer.send(new ProducerRecord<>("topic", 1, "key", "value")); }