From ee34b489f630c9c44eb9d9853d1eda67023d3055 Mon Sep 17 00:00:00 2001 From: buildmaster Date: Thu, 7 May 2020 05:29:16 +0000 Subject: [PATCH] Bumping versions --- .../SleuthKafkaStreamsConfiguration.java | 31 ++++++++------ ...aStreamsConfigurationIntegrationTests.java | 40 +++++++++++-------- 2 files changed, 42 insertions(+), 29 deletions(-) diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfiguration.java index 32e0e6dc5..8570db8d9 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfiguration.java @@ -45,7 +45,8 @@ import org.springframework.kafka.config.StreamsBuilderFactoryBean; @ConditionalOnBean(Tracing.class) @AutoConfigureAfter({ TraceAutoConfiguration.class }) @OnMessagingEnabled -@ConditionalOnProperty(value = "spring.sleuth.messaging.kafka.streams.enabled", matchIfMissing = true) +@ConditionalOnProperty(value = "spring.sleuth.messaging.kafka.streams.enabled", + matchIfMissing = true) @ConditionalOnClass(KafkaStreams.class) public class SleuthKafkaStreamsConfiguration { @@ -55,7 +56,6 @@ public class SleuthKafkaStreamsConfiguration { /** * Expose {@link KafkaStreamsTracing} as bean to allow for filter/map/peek/transform * operations. - * * @param tracing Brave Tracing instance from TraceAutoConfiguration * @return instance for use in further manual instrumentation */ @@ -66,9 +66,9 @@ public class SleuthKafkaStreamsConfiguration { } /** - * Call {@link StreamsBuilderFactoryBean#setClientSupplier(org.apache.kafka.streams.KafkaClientSupplier)} with - * Brave's TracingKafkaClientSupplier. - * + * Call + * {@link StreamsBuilderFactoryBean#setClientSupplier(org.apache.kafka.streams.KafkaClientSupplier)} + * with Brave's TracingKafkaClientSupplier. * @param objectProvider provides KafkaStreamsTracing; prevents eager initialization * @return */ @@ -91,22 +91,29 @@ public class SleuthKafkaStreamsConfiguration { */ class KafkaStreamsBuilderFactoryBeanPostProcessor implements BeanPostProcessor { - private static final Log log = LogFactory.getLog(KafkaStreamsBuilderFactoryBeanPostProcessor.class); + private static final Log log = LogFactory + .getLog(KafkaStreamsBuilderFactoryBeanPostProcessor.class); private final ObjectProvider objectProvider; - KafkaStreamsBuilderFactoryBeanPostProcessor(ObjectProvider objectProvider) { + KafkaStreamsBuilderFactoryBeanPostProcessor( + ObjectProvider objectProvider) { this.objectProvider = objectProvider; } @Override - public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + public Object postProcessAfterInitialization(Object bean, String beanName) + throws BeansException { if (bean instanceof StreamsBuilderFactoryBean) { - // KafkaStreamsTracing is created in SleuthKafkaStreamsConfiguration above, so should not be null here - KafkaStreamsTracing kafkaStreamsTracing = this.objectProvider.getIfAvailable(); - ((StreamsBuilderFactoryBean) bean).setClientSupplier(kafkaStreamsTracing.kafkaClientSupplier()); + // KafkaStreamsTracing is created in SleuthKafkaStreamsConfiguration above, so + // should not be null here + KafkaStreamsTracing kafkaStreamsTracing = this.objectProvider + .getIfAvailable(); + ((StreamsBuilderFactoryBean) bean) + .setClientSupplier(kafkaStreamsTracing.kafkaClientSupplier()); if (log.isDebugEnabled()) { - log.debug("StreamsBuilderFactoryBean bean is auto-configured to enable tracing."); + log.debug( + "StreamsBuilderFactoryBean bean is auto-configured to enable tracing."); } } return bean; diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfigurationIntegrationTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfigurationIntegrationTests.java index 77cc0a07a..5aba918e5 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfigurationIntegrationTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthKafkaStreamsConfigurationIntegrationTests.java @@ -46,60 +46,60 @@ import static org.mockito.Mockito.verify; class SleuthKafkaStreamsConfigurationIntegrationTests { private final ApplicationContextRunner contextRunner = new ApplicationContextRunner() - .withConfiguration(AutoConfigurations.of( - TraceAutoConfiguration.class, + .withConfiguration(AutoConfigurations.of(TraceAutoConfiguration.class, SleuthKafkaStreamsConfiguration.class)) .withUserConfiguration(UserConfig.class); @Test void should_create_KafkaStreamsTracing() { - this.contextRunner - .run(context -> assertThat(context).hasSingleBean(KafkaStreamsTracing.class)); + this.contextRunner.run( + context -> assertThat(context).hasSingleBean(KafkaStreamsTracing.class)); } @Test void should_not_create_KafkaStreamsTracing_when_KafkaStreams_not_present() { - this.contextRunner - .withClassLoader(new FilteredClassLoader(KafkaStreams.class)) - .run(context -> assertThat(context).doesNotHaveBean(KafkaStreamsTracing.class)); + this.contextRunner.withClassLoader(new FilteredClassLoader(KafkaStreams.class)) + .run(context -> assertThat(context) + .doesNotHaveBean(KafkaStreamsTracing.class)); } @Test void should_not_create_KafkaStreamsTracing_when_kafkastreams_disabled() { this.contextRunner .withPropertyValues("spring.sleuth.messaging.kafka.streams.enabled=false") - .run(context -> assertThat(context).doesNotHaveBean(KafkaStreamsTracing.class)); + .run(context -> assertThat(context) + .doesNotHaveBean(KafkaStreamsTracing.class)); } @Test void should_not_create_KafkaStreamsTracing_when_messaging_disabled() { - this.contextRunner - .withPropertyValues("spring.sleuth.messaging.enabled=false") - .run(context -> assertThat(context).doesNotHaveBean(KafkaStreamsTracing.class)); + this.contextRunner.withPropertyValues("spring.sleuth.messaging.enabled=false") + .run(context -> assertThat(context) + .doesNotHaveBean(KafkaStreamsTracing.class)); } @Test void should_set_KafkaClientSupplier_on_StreamsBuilderFactoryBean() { - this.contextRunner - .run(context -> verify(UserConfig.streamsBuilderFactoryBean) - .setClientSupplier(any(KafkaClientSupplier.class))); + this.contextRunner.run(context -> verify(UserConfig.streamsBuilderFactoryBean) + .setClientSupplier(any(KafkaClientSupplier.class))); } @Test void should_not_complain_about_eager_initialization() { - this.contextRunner - .withUserConfiguration(EagerInitializationConfig.class) + this.contextRunner.withUserConfiguration(EagerInitializationConfig.class) .run(context -> verify(UserConfig.streamsBuilderFactoryBean) .setClientSupplier(any(KafkaClientSupplier.class))); } @AfterEach void afterEach(CapturedOutput output) { - assertThat(output).doesNotContain("is not eligible for getting processed by all BeanPostProcessors"); + assertThat(output).doesNotContain( + "is not eligible for getting processed by all BeanPostProcessors"); } @Configuration static class UserConfig { + static StreamsBuilderFactoryBean streamsBuilderFactoryBean; @Bean @@ -107,25 +107,31 @@ class SleuthKafkaStreamsConfigurationIntegrationTests { streamsBuilderFactoryBean = mock(StreamsBuilderFactoryBean.class); return UserConfig.streamsBuilderFactoryBean; } + } @Configuration static class EagerInitializationConfig { + @Bean EagerInitializationComponent eagerInitializationComponent() { return new EagerInitializationComponent(); } + } static class EagerInitializationComponent { @Autowired private Tracing tracing; + private KafkaStreamsTracing kafkaStreamsTracing; @PostConstruct void init() { kafkaStreamsTracing = KafkaStreamsTracing.create(tracing); } + } + }