Spring Boot: Add configuration to reference a subscription by ID (#440)

* feat(config): add subscription id as a config property

* feat(config): update readme with subscription config

* feat(config): revert README.md styling

* feat(config): change subscription-id to subscription-by-id

* feat(config): fix README.md

* feat(config): add default string for subscription by id

* Minor grammar improvement in README

---------

Co-authored-by: Kunal Mangaraj <[email protected]>

fahrschein-spring-boot-starter/src/main/java/org/zalando/spring/boot/fahrschein/nakadi/config/FahrscheinNakadiConsumer.java

@@ -130,19 +130,25 @@ private Subscription getSubscription() throws IOException {
         }
 
         SubscriptionBuilder sb = nakadiClient
-                .subscription(consumerConfig.getApplicationName(), new HashSet<>(consumerConfig.getTopics()))
-                .withConsumerGroup(consumerConfig.getConsumerGroup())
-                .withAuthorization(authorization()
-                        .withAdmins(adminAttributes)
-                        .withReaders(readerAttributes)
-                        .build());
+                .subscription(consumerConfig.getApplicationName(), new HashSet<>(consumerConfig.getTopics()));
 
         if (END.equals(consumerConfig.getReadFrom())) {
             sb = sb.readFromEnd();
         } else {
             sb = sb.readFromBegin();
         }
-        return sb.subscribe();
+
+        if (consumerConfig.getSubscriptionById() != null && !consumerConfig.getSubscriptionById().isEmpty()) {
+            return sb.subscribe(consumerConfig.getSubscriptionById());
+        } else {
+            sb
+                .withConsumerGroup(consumerConfig.getConsumerGroup())
+                .withAuthorization(authorization()
+                        .withAdmins(adminAttributes)
+                        .withReaders(readerAttributes)
+                        .build());
+            return sb.subscribe();
+        }
     }
 
     protected StreamParameters getStreamParameters() {

fahrschein-spring-boot-starter/src/main/java/org/zalando/spring/boot/fahrschein/nakadi/config/properties/AbstractConfig.java

@@ -24,6 +24,8 @@ public abstract class AbstractConfig {
 
     private String objectMapperRef;
 
+    private String subscriptionById;
+
     @NestedConfigurationProperty
     protected OAuthConfig oauth = OAuthConfig.defaultOAuthConfig();
 
@@ -37,5 +39,4 @@ public abstract class AbstractConfig {
     protected BackoffConfig backoff = BackoffConfig.defaultBackoffConfig();
 
     protected ThreadConfig threads = new ThreadConfig();
-
 }

fahrschein-spring-boot-starter/src/main/java/org/zalando/spring/boot/fahrschein/nakadi/config/properties/ConsumerConfig.java

@@ -28,6 +28,7 @@ void mergeWithDefaultConfig(DefaultConsumerConfig defaultConsumerConfig) {
         this.setReadFrom(ofNullable(this.getReadFrom()).orElse(defaultConsumerConfig.getReadFrom()));
         this.setRecordMetrics(ofNullable(this.getRecordMetrics()).orElse(defaultConsumerConfig.getRecordMetrics()));
         this.setObjectMapperRef(ofNullable(this.getObjectMapperRef()).orElse(defaultConsumerConfig.getObjectMapperRef()));
+        this.setSubscriptionById(merge(this.getSubscriptionById(), defaultConsumerConfig.getSubscriptionById()));
 
         // oauth
         if(defaultConsumerConfig.getOauth().getEnabled() && !this.getOauth().getEnabled()) {

fahrschein-spring-boot-starter/src/main/java/org/zalando/spring/boot/fahrschein/nakadi/config/properties/DefaultConsumerConfig.java

@@ -20,6 +20,7 @@ public DefaultConsumerConfig() {
         setReadFrom(Position.END);
         setRecordMetrics(Boolean.FALSE);
         setObjectMapperRef(FAHRSCHEIN_OBJECT_MAPPER_REF_NAME);
+        setSubscriptionById("");
     }
 
 }

fahrschein-spring-boot-starter/src/test/java/org/zalando/spring/boot/fahrschein/nakadi/ApplicationTest.java

@@ -139,6 +139,7 @@ public void hasProperlyDeserializedConfigProperties() {
                 .hasFieldOrPropertyWithValue("nakadiUrl", dc.getNakadiUrl())
                 .hasFieldOrPropertyWithValue("consumerGroup", dc.getConsumerGroup())
                 .hasFieldOrPropertyWithValue("readFrom", dc.getReadFrom())
+                .hasFieldOrPropertyWithValue("subscriptionById", "test-by-id")
                 .hasNoNullFieldsOrProperties();
 
         // consumer oauth config

fahrschein-spring-boot-starter/src/test/java/org/zalando/spring/boot/fahrschein/nakadi/config/properties/ConsumerConfigTest.java

@@ -24,6 +24,7 @@ public class ConsumerConfigTest {
     private static final String ACCESS_TOKEN_ID = "testAccessTokenId";
     private static final String NAKADI_URI = "http://localhost";
     private static final String APPLICATION_NAME = "TestApp";
+    private static final String SUBSCRIPTION_BY_ID = "test-by-id";
 
     @Test
     public void testInitialConsumerConfig() {
@@ -108,6 +109,7 @@ public void testMergedConsumerConfig() {
         assertThat(cc.getStreamParameters().getStreamTimeout()).isEqualTo(STREAM_TIMEOUT);
 
         assertThat(cc.getHttp().getContentEncoding()).isEqualTo(ContentEncoding.GZIP);
+        assertThat(cc.getSubscriptionById()).isEqualTo(SUBSCRIPTION_BY_ID);
     }
 
     private DefaultConsumerConfig getDefaultConsumerConfig() {
@@ -135,6 +137,7 @@ private DefaultConsumerConfig getDefaultConsumerConfig() {
         dc.getStreamParameters().setStreamKeepAliveLimit(STREAM_KEEP_ALIVE_LIMIT);
         dc.getStreamParameters().setStreamLimit(STREAM_LIMIT);
         dc.getStreamParameters().setStreamTimeout(STREAM_TIMEOUT);
+        dc.setSubscriptionById(SUBSCRIPTION_BY_ID);
 
         return dc;
     }

fahrschein-spring-boot-starter/src/test/resources/application.yml

@@ -50,3 +50,4 @@ fahrschein:
         content-encoding: identity
       topics:
       - first.first-update
+      subscription-by-id: "test-by-id"