Introduce "switch off all submission" toggle

- auto configuration
- test for non-existence of certain beans
- documentation

README.md

@@ -163,6 +163,23 @@ If you do not use the STUPS Tokens library, you can implement token retrieval yo
 type [`AccessTokenProvider`](nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/AccessTokenProvider.java).
 The starter will detect it and call it once for each request to retrieve the token.
 
+### Disable submission completely
+
+You can disable the whole Nakadi integration completely with this property:
+
+```yaml
+nakadi-producer:
+   submission-enabled: false
+```
+
+In this case you don't need to configure anything related to the Nakadi communication ↑ (and this library won't
+set up any beans related to it).
+
+A use case for this might be that you have several components of your application connected to the same database,
+and want the submission of the events centralized in one of these components. Then for all other components you'd
+set `nakadi-producer.submission-enabled: false` (true is the default), but still can use the EventLogWriter to
+create events.
+
 ### Creating events
 
 The typical use case for this library is to publish events like creating or updating of some objects.
@@ -431,7 +448,8 @@ This is a list of all the documented spring properties (in alphabetical order),
 | [`nakadi-producer.lock-duration-buffer`](#customizing-event-locks) | Number of seconds before the expiry of a lock an event is not used. |
 | [`nakadi-producer.lock-size`](#customizing-event-locks)          | Number of events to lock (and then load into memory) at once.  |
 | [`nakadi-producer.nakadi-base-uri`](#letting-this-library-set-things-up) | The Nakadi base URI used for submitting events.  |
-| [`nakadi-producer.scheduled-transmission-enabled: false`](#test-support) | Disable event transmission scheduler.  |
+| [`nakadi-producer.scheduled-transmission-enabled: false`](#test-support) | Disable event transmission scheduler (but still set up Nakadi connection beans, so it can be used manually). |
+| [`nakadi-producer.submission-enabled: false`](#disable-submission-completely) | Disable event submission completely (including all beans for this). |
 | [`tracer.traces.X-Flow-ID: flow-id`](#x-flow-id-optional)        | Enable flow-ID support |
 
 ## Contributing

nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java

@@ -14,6 +14,7 @@
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.flyway.FlywayProperties;
 import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -47,24 +48,27 @@
 
 @Configuration
 @AutoConfigureAfter(name="org.zalando.tracer.spring.TracerAutoConfiguration")
-@EnableScheduling
 @EnableConfigurationProperties({ DataSourceProperties.class, FlywayProperties.class })
 @Slf4j
 public class NakadiProducerAutoConfiguration {
 
+    @ConditionalOnProperty(name="nakadi-producer.submission-enabled", havingValue = "true", matchIfMissing = true)
     @ConditionalOnMissingBean({NakadiPublishingClient.class, NakadiClient.class})
     @Configuration
     @Import(FahrscheinWithTokensNakadiClientConfiguration.StupsTokenConfiguration.class)
     static class FahrscheinWithTokensNakadiClientConfiguration {
 
         @Bean
-        public NakadiPublishingClient nakadiProducerPublishingClient(AccessTokenProvider accessTokenProvider,
-                @Value("${nakadi-producer.nakadi-base-uri}") URI nakadiBaseUri, RequestFactory requestFactory) {
+        public NakadiPublishingClient nakadiProducerPublishingClient(
+                AccessTokenProvider accessTokenProvider,
+                @Value("${nakadi-producer.nakadi-base-uri}") URI nakadiBaseUri,
+                RequestFactory requestFactory) {
             return new FahrscheinNakadiPublishingClient(NakadiClient.builder(nakadiBaseUri, requestFactory)
                     .withAccessTokenProvider(accessTokenProvider::getAccessToken).build());
         }
 
         @ConditionalOnClass(name = "org.zalando.stups.tokens.Tokens")
+        @ConditionalOnProperty(name="nakadi-producer.submission-enabled", havingValue = "true", matchIfMissing = true)
         @ConditionalOnMissingBean({NakadiPublishingClient.class, NakadiClient.class})
         @Configuration
         static class StupsTokenConfiguration {
@@ -75,16 +79,16 @@ public StupsTokenComponent accessTokenProvider(
                     @Value("${nakadi-producer.access-token-scopes:uid}") String[] accessTokenScopes) {
                 return new StupsTokenComponent(accessTokenUri, Arrays.asList(accessTokenScopes));
             }
-
         }
+
         @Bean
         @ConditionalOnMissingBean
-        RequestFactory requestFactory(@Value("${nakadi-producer.encoding:GZIP}") ContentEncoding encoding){
+        RequestFactory requestFactory(@Value("${nakadi-producer.encoding:GZIP}") ContentEncoding encoding) {
             return new SimpleRequestFactory(encoding);
         }
-
     }
 
+    @ConditionalOnProperty(name="nakadi-producer.submission-enabled", havingValue = "true", matchIfMissing = true)
     @ConditionalOnMissingBean(NakadiPublishingClient.class)
     @ConditionalOnBean(NakadiClient.class)
     @Configuration
@@ -150,27 +154,34 @@ public EventLogRepository eventLogRepository(NamedParameterJdbcTemplate namedPar
         return new EventLogRepositoryImpl(namedParameterJdbcTemplate, lockSize);
     }
 
-    @Bean
-    public EventTransmitter eventTransmitter(EventTransmissionService eventTransmissionService) {
-        return new EventTransmitter(eventTransmissionService);
-    }
+    @ConditionalOnProperty(name="nakadi-producer.submission-enabled", havingValue = "true", matchIfMissing = true)
+    @EnableScheduling
+    @Configuration
+    static class TransmissionConfiguration {
 
-    @Bean
-    public EventTransmissionScheduler eventTransmissionScheduler(EventTransmitter eventTransmitter,
-            @Value("${nakadi-producer.scheduled-transmission-enabled:true}") boolean scheduledTransmissionEnabled) {
-        return new EventTransmissionScheduler(eventTransmitter, scheduledTransmissionEnabled);
-    }
+        @Bean
+        public EventTransmitter eventTransmitter(EventTransmissionService eventTransmissionService) {
+            return new EventTransmitter(eventTransmissionService);
+        }
+
+        @Bean
+        public EventTransmissionScheduler eventTransmissionScheduler(
+                EventTransmitter eventTransmitter,
+                @Value("${nakadi-producer.scheduled-transmission-enabled:true}") boolean scheduledTransmissionEnabled) {
+            return new EventTransmissionScheduler(eventTransmitter, scheduledTransmissionEnabled);
+        }
 
-  @Bean
-  public EventTransmissionService eventTransmissionService(
-      EventLogRepository eventLogRepository,
-      NakadiPublishingClient nakadiPublishingClient,
-      ObjectMapper objectMapper,
-      @Value("${nakadi-producer.lock-duration:600}") int lockDuration,
-      @Value("${nakadi-producer.lock-duration-buffer:60}") int lockDurationBuffer) {
-    return new EventTransmissionService(
-        eventLogRepository, nakadiPublishingClient, objectMapper, lockDuration, lockDurationBuffer);
-  }
+        @Bean
+        public EventTransmissionService eventTransmissionService(
+                EventLogRepository eventLogRepository,
+                NakadiPublishingClient nakadiPublishingClient,
+                ObjectMapper objectMapper,
+                @Value("${nakadi-producer.lock-duration:600}") int lockDuration,
+                @Value("${nakadi-producer.lock-duration-buffer:60}") int lockDurationBuffer) {
+            return new EventTransmissionService(
+                    eventLogRepository, nakadiPublishingClient, objectMapper, lockDuration, lockDurationBuffer);
+        }
+    }
 
     @Bean
     public FlywayMigrator flywayMigrator() {

nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/SubmissionDisabledIT.java

@@ -0,0 +1,62 @@
+package org.zalando.nakadiproducer;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.ApplicationContext;
+import org.springframework.test.context.ActiveProfiles;
+import org.zalando.fahrschein.NakadiClient;
+import org.zalando.fahrschein.http.api.RequestFactory;
+import org.zalando.nakadiproducer.config.EmbeddedDataSourceConfig;
+import org.zalando.nakadiproducer.eventlog.EventLogWriter;
+import org.zalando.nakadiproducer.eventlog.impl.EventLogRepository;
+import org.zalando.nakadiproducer.transmission.NakadiPublishingClient;
+import org.zalando.nakadiproducer.transmission.impl.EventTransmissionService;
+import org.zalando.nakadiproducer.transmission.impl.EventTransmitter;
+
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+// no "test" profile, as this would include the mock client.
+@SpringBootTest(
+        webEnvironment = SpringBootTest.WebEnvironment.MOCK,
+        properties = {"nakadi-producer.submission-enabled:false"},
+        classes = { TestApplication.class, EmbeddedDataSourceConfig.class }
+)
+public class SubmissionDisabledIT  {
+
+    @Autowired
+    ApplicationContext context;
+
+    @Test
+    public void noNakadiBeans() {
+        assertThat(context.getBeanProvider(NakadiClient.class).getIfAvailable(), nullValue());
+        assertThat(context.getBeanProvider(NakadiPublishingClient.class).getIfAvailable(), nullValue());
+        assertThat(context.getBeanProvider(StupsTokenComponent.class).getIfAvailable(), nullValue());
+        assertThat(context.getBeanProvider(RequestFactory.class).getIfAvailable(), nullValue());
+    }
+
+    @Test
+    public void noTransmissionBeans() {
+        assertThat(context.getBeanProvider(EventTransmitter.class).getIfAvailable(), nullValue());
+        assertThat(context.getBeanProvider(EventTransmissionService.class).getIfAvailable(), nullValue());
+        assertThat(context.getBeanProvider(EventTransmissionScheduler.class).getIfAvailable(), nullValue());
+    }
+
+    @Test
+    public void yesEventLogWriter() {
+        assertThat(context.getBeanProvider(EventLogWriter.class).getIfAvailable(), notNullValue());
+    }
+
+    @Test
+    public void yesRepository() {
+        assertThat(context.getBeanProvider(EventLogRepository.class).getIfAvailable(), notNullValue());
+    }
+
+    @Test
+    public void yesFlywayMigrator() {
+        assertThat(context.getBeanProvider(FlywayMigrator.class).getIfAvailable(), notNullValue());
+    }
+
+}