### Question:
StreamableKafkaMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { public static <K, V> Builder<K, V> builder() { return new Builder<>(); } protected  StreamableKafkaMessageSource(Builder<K, V> builder); static Builder<K, V> builder(); @Override BlockingStream<TrackedEventMessage<?>> openStream(TrackingToken trackingToken);  }### Answer:
@Test void testBuildingWithInvalidTopicsShouldThrowAxonConfigurationException() { assertThrows(AxonConfigurationException.class, () -> StreamableKafkaMessageSource.builder().topics(null)); }
@Test void testBuildWithInvalidTopicThrowsAxonConfigurationException() { assertThrows(AxonConfigurationException.class, () -> StreamableKafkaMessageSource.builder().addTopic(null)); }
@Test void testBuildingWithInvalidGroupIdPrefixShouldThrowAxonConfigurationException() { assertThrows( AxonConfigurationException.class, () -> StreamableKafkaMessageSource.builder().groupIdPrefix(null) ); }
@Test void testBuildingWithInvalidGroupIdSuffixFactoryShouldThrowAxonConfigurationException() { assertThrows( AxonConfigurationException.class, () -> StreamableKafkaMessageSource.builder().groupIdSuffixFactory(null) ); }
@Test void testBuildingWithInvalidConsumerFactoryShouldThrowAxonConfigurationException() { assertThrows( AxonConfigurationException.class, () -> StreamableKafkaMessageSource.builder().consumerFactory((ConsumerFactory) null) ); }
@Test void testBuildingWithInvalidFetcherShouldThrowAxonConfigurationException() { assertThrows(AxonConfigurationException.class, () -> StreamableKafkaMessageSource.builder().fetcher(null)); }
@Test void testBuildingWithInvalidMessageConverterShouldThrowAxonConfigurationException() { assertThrows( AxonConfigurationException.class, () -> StreamableKafkaMessageSource.builder().messageConverter(null) ); }
@Test void testBuildingWithInvalidBufferFactoryShouldThrowAxonConfigurationException() { assertThrows( AxonConfigurationException.class, () -> StreamableKafkaMessageSource.builder().bufferFactory(null) ); }
@Test void testBuildingWhilstMissingRequiredFieldsShouldThrowAxonConfigurationException() { assertThrows(AxonConfigurationException.class, () -> StreamableKafkaMessageSource.builder().build()); }