### Question:
SubscribableKafkaMessageSource implements SubscribableMessageSource<EventMessage<?>> { public static <K, V> Builder<K, V> builder() { return new Builder<>(); } @SuppressWarnings("WeakerAccess") protected  SubscribableKafkaMessageSource(Builder<K, V> builder); static Builder<K, V> builder(); @Override Registration subscribe(java.util.function.Consumer<List<? extends EventMessage<?>>> eventProcessor); void start(); void close();  }### Answer:
@Test void testBuildWithInvalidTopicsThrowsAxonConfigurationException() { assertThrows(AxonConfigurationException.class, () -> SubscribableKafkaMessageSource.builder().topics(null)); }
@Test void testBuildWithInvalidTopicThrowsAxonConfigurationException() { assertThrows(AxonConfigurationException.class, () -> SubscribableKafkaMessageSource.builder().addTopic(null)); }
@Test void testBuildWithInvalidGroupIdThrowsAxonConfigurationException() { assertThrows(AxonConfigurationException.class, () -> SubscribableKafkaMessageSource.builder().groupId(null)); }
@Test void testBuildWithInvalidConsumerFactoryThrowsAxonConfigurationException() { assertThrows( AxonConfigurationException.class, () -> SubscribableKafkaMessageSource.builder().consumerFactory((ConsumerFactory) null) ); }
@Test void testBuildWithInvalidFetcherThrowsAxonConfigurationException() { assertThrows(AxonConfigurationException.class, () -> SubscribableKafkaMessageSource.builder().fetcher(null)); }
@Test void testBuildWithInvalidMessageConverterThrowsAxonConfigurationException() { assertThrows( AxonConfigurationException.class, () -> SubscribableKafkaMessageSource.builder().messageConverter(null) ); }
@Test void testBuildWithInvalidConsumerCountThrowsAxonConfigurationException() { assertThrows( AxonConfigurationException.class, () -> SubscribableKafkaMessageSource.builder().consumerCount(0) ); }
@Test void testBuildingWhilstMissingRequiredFieldsShouldThrowAxonConfigurationException() { assertThrows(AxonConfigurationException.class, () -> SubscribableKafkaMessageSource.builder().build()); }