Support spring boot 2

README.md

@@ -16,7 +16,7 @@ The goal of this Spring Boot starter is to simplify the reliable integration bet
 
 There are already [multiple clients for the Nakadi REST API](https://zalando.github.io/nakadi/manual.html#using_clients), but none of them solves the mentioned issues. 
 
-We solved them by persisting new events in a log table as part of the producing JDBC transaction. They will then be sent asynchonously to Nakadi after the transaction completed. If the transaction is rolled back, the events will vanish too. As a result, events will always be sent if and only if the transaction succeeded.
+We solved them by persisting new events in a log table as part of the producing JDBC transaction. They will then be sent asynchronously to Nakadi after the transaction completed. If the transaction is rolled back, the events will vanish too. As a result, events will always be sent if and only if the transaction succeeded.
 
 The Transmitter generates a strictly monotonically increasing event id that can be used for ordering the events during retrieval. It is not guaranteed, that events will be sent to Nakadi in the order they have been produced. If an event could not be sent to Nakadi, the library will periodically retry the transmission.
 
@@ -40,7 +40,8 @@ You may of course always setup a fresh system with the newest version.
 
 ## Prerequisites
 
-This library was tested with Spring Boot 1.5.3.RELEASE and relies on an existing configured PostgreSQL DataSource.
+This library was tested with Spring Boot 2.0.3.RELEASE and relies on an existing configured PostgreSQL DataSource. 
+**If you are still using Spring Boot 1.x, please use versions < 20.0.0, they are still actively maintained ([Documentation](https://github.com/zalando-nakadi/nakadi-producer-spring-boot-starter/tree/spring-boot-1)).**
 
 This library also uses:
 
@@ -50,14 +51,14 @@ This library also uses:
 * jackson >= 2.7.0
 * (optional) Zalando's [tracer-spring-boot-starter](https://github.com/zalando/tracer)
 * (optional) Zalando's [tokens library](https://github.com/zalando/tokens) >= 0.10.0
-    * Please note that [tokens-spring-boot-starter](https://github.com/zalando-stups/spring-boot-zalando-stups-tokens) 0.10.0 comes with tokens 0.9.9, which is not enough. You can manually add tokens 0.10.0 with that starter, though.
+    * Please note that [tokens-spring-boot-starter](https://github.com/zalando-stups/spring-boot-zalando-stups-tokens) 0.10.0 comes with tokens 0.9.9, which is not enough. You can manually add tokens 0.10.0 with that starter, though. To be used in zalando's k8s environment, you must at least use 0.11.0.
 
 
 ## Usage
 
 ### Setup
 
-If you are using maven, include the library in your `pom.xml`:
+If you are using Maven, include the library in your `pom.xml`:
 
 ```xml
 <dependency>
@@ -81,7 +82,7 @@ public class Application {
 }
 ```
 
-The library uses flyway migrations to set up its own database schema `nakadi_events`.
+The library uses Flyway migrations to set up its own database schema `nakadi_events`.
 
 ### Nakadi communication configuration
 
@@ -211,30 +212,43 @@ process step the event is reporting.
 ### Event snapshots (optional)
 
 A Snapshot event is a special type of data change event (data operation) defined by Nakadi.
-It does not represent a change of the state of a resource, but a current snapshot of the state of the resource.
+It does not represent a change of the state of a resource, but a current snapshot of its state. It can be useful to
+bootstrap a new consumer or to recover from inconsistencies between sender and consumer after an incident.
 
 You can create snapshot events programmatically (using EventLogWriter.fireSnapshotEvent), but usually snapshot event
 creation is a irregular, manually triggered maintenance task.
 
 This library provides a Spring Boot Actuator endpoint named `snapshot_event_creation` that can be used to trigger a Snapshot for a given event type. Assuming your management port is set to `7979`,
 
-    GET localhost:7979/snapshot_event_creation
+    GET localhost:7979/actuator/snapshot-event-creation
 
 will return a list of all event types available for snapshot creation and 
 
-    POST localhost:7979/snapshot_event_creation/my.event-type
+    POST localhost:7979/actuator/snapshot-event-creation/my.event-type
 
-will trigger a snapshot for the event type `my.event-type`. The (optional) request body is a "filter specifier".
+will trigger a snapshot for the event type `my.event-type`. You can change the port, the authentication scheme and the
+path prefix as part of your Spring Boot Actuator configuration.
 
-This will only  work if your application has configured spring-boot-actuator
+You can provide an optional filter specifier that will be passed to your application to implement any application 
+specific event/entity filtering logic.  It can be provided either as a query parameter called `filter`, or as a
+request body
+
+    {"filter":"myFilter"}
+
+This endpoint will only work if your application includes spring-boot-actuator,
 ```xml
 <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-actuator</artifactId>
 </dependency>
 ```
-and if one or more Spring Beans implement the `org.zalando.nakadiproducer.snapshots.SnapshotEventGenerator` interface. Otherwise (or if the generator is not for the event type you requested), the library will respond with an error message when you request a snapshot creation.
-The request body (the "filter specifier") of the trigger request will be passed as a string parameter to the SnapshotEventGenerator's `generateSnapshots` method.
+your `application.properties` includes
+``` 
+management.endpoints.web.exposure.include=snapshot-event-creation,your-other-endpoints,...`
+```
+and if one or more Spring Beans implement the `org.zalando.nakadiproducer.snapshots.SnapshotEventGenerator` interface.
+The optional filter specifier of the trigger request will be passed as a string parameter to the
+SnapshotEventGenerator's `generateSnapshots` method and may be null, if none is given.
 
 We provide a `SimpleSnapshotEventGenerator` to ease bean creation using a more functional style:
 ```java
@@ -269,13 +283,13 @@ tracer:
 By default, the library will pick up your flyway data source (or the primary data source if no flyway data source is
 configured), create its own schema and start setting up its tables in there. You can customize this process in two ways:
 
-If you want to use a different data source for schema maintainence (for example to use a different username) and 
-configuring the Spring flyway datasource is not enough, your can define a spring bean of type `DataSource` and annotate 
+If you want to use a different data source for schema maintenance (for example to use a different username) and 
+configuring the Spring Flyway datasource is not enough, your can define a spring bean of type `DataSource` and annotate 
 it with `@NakadiProducerDataSource`.
 
-You may also define a spring bean of type `FlywayCallback` and annotate it with `@NakadiProducerFlywayCallback`. The
-interface provide several hook into the schema management lifecycle that may, for example, be used to
- `SET ROLE migrator` before and `RESET ROLE` after each migration. 
+You may also define a spring bean of type `NakadiProducerFlywayCallback`. The interface provides several hooks into the
+schema management lifecycle that may, for example, be used to `SET ROLE migrator` before and `RESET ROLE` after each
+migration. 
 
 ### Test support
 

nakadi-producer-spring-boot-starter/pom.xml

@@ -10,7 +10,7 @@
     <parent>
         <groupId>org.zalando</groupId>
         <artifactId>nakadi-producer-reactor</artifactId>
-        <version>4.2.0</version>
+        <version>20.0.0-SNAPSHOT</version>
     </parent>
 
     <artifactId>nakadi-producer-spring-boot-starter</artifactId>
@@ -102,6 +102,14 @@
             <artifactId>spring-boot-starter-web</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-actuator-autoconfigure</artifactId>
+        </dependency>
     </dependencies>
 
 

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

@@ -4,11 +4,18 @@
 import javax.sql.DataSource;
 
 import org.flywaydb.core.Flyway;
+import org.flywaydb.core.api.MigrationInfo;
+import org.flywaydb.core.api.callback.BaseFlywayCallback;
 import org.flywaydb.core.api.callback.FlywayCallback;
+import org.flywaydb.core.api.configuration.ConfigurationAware;
+import org.flywaydb.core.api.configuration.FlywayConfiguration;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.flyway.FlywayDataSource;
 import org.springframework.boot.autoconfigure.flyway.FlywayProperties;
 
+import java.sql.Connection;
+import java.util.List;
+
 public class FlywayMigrator {
     @Autowired(required = false)
     @NakadiProducerFlywayDataSource
@@ -22,10 +29,9 @@ public class FlywayMigrator {
     private DataSource dataSource;
 
     @Autowired(required = false)
-    @NakadiProducerFlywayCallback
-    private FlywayCallback callback;
+    private List<NakadiProducerFlywayCallback> callbacks;
 
-    @Autowired
+    @Autowired(required = false)
     private FlywayProperties flywayProperties;
 
     @PostConstruct
@@ -34,7 +40,7 @@ public void migrateFlyway() {
 
         if (this.nakadiProducerFlywayDataSource != null) {
             flyway.setDataSource(nakadiProducerFlywayDataSource);
-        } else if (this.flywayProperties.isCreateDataSource()) {
+        } else if (this.flywayProperties != null && this.flywayProperties.isCreateDataSource()) {
             flyway.setDataSource(this.flywayProperties.getUrl(), this.flywayProperties.getUser(),
                 this.flywayProperties.getPassword(),
                 this.flywayProperties.getInitSqls().toArray(new String[0]));
@@ -46,11 +52,118 @@ public void migrateFlyway() {
 
         flyway.setLocations("classpath:db_nakadiproducer/migrations");
         flyway.setSchemas("nakadi_events");
-        if (callback != null) {
-            flyway.setCallbacks(callback);
+        if (callbacks != null) {
+            flyway.setCallbacks(callbacks.stream().map(FlywayCallbackAdapter::new).toArray(FlywayCallback[]::new));
         }
+
         flyway.setBaselineOnMigrate(true);
         flyway.setBaselineVersionAsString("2133546886.1.0");
         flyway.migrate();
     }
+
+    private static class FlywayCallbackAdapter extends BaseFlywayCallback {
+
+        private NakadiProducerFlywayCallback callback;
+
+        private FlywayCallbackAdapter(NakadiProducerFlywayCallback callback) {
+            this.callback = callback;
+        }
+
+        @Override
+        public void setFlywayConfiguration(FlywayConfiguration flywayConfiguration) {
+            if (callback instanceof ConfigurationAware) {
+                ((ConfigurationAware) callback).setFlywayConfiguration(flywayConfiguration);
+            }
+        }
+
+        @Override
+        public void beforeClean(Connection connection) {
+            callback.beforeClean(connection);
+        }
+
+        @Override
+        public void afterClean(Connection connection) {
+            callback.afterClean(connection);
+        }
+
+        @Override
+        public void beforeMigrate(Connection connection) {
+            callback.beforeMigrate(connection);
+        }
+
+        @Override
+        public void afterMigrate(Connection connection) {
+            callback.afterMigrate(connection);
+        }
+
+        @Override
+        public void beforeEachMigrate(Connection connection, MigrationInfo info) {
+            callback.beforeEachMigrate(connection, info);
+        }
+
+        @Override
+        public void afterEachMigrate(Connection connection, MigrationInfo info) {
+            callback.afterEachMigrate(connection, info);
+        }
+
+        @Override
+        public void beforeUndo(Connection connection) {
+            callback.beforeUndo(connection);
+        }
+
+        @Override
+        public void beforeEachUndo(Connection connection, MigrationInfo info) {
+            callback.beforeEachUndo(connection, info);
+        }
+
+        @Override
+        public void afterEachUndo(Connection connection, MigrationInfo info) {
+            callback.afterEachUndo(connection, info);
+        }
+
+        @Override
+        public void afterUndo(Connection connection) {
+            callback.afterUndo(connection);
+        }
+
+        @Override
+        public void beforeValidate(Connection connection) {
+            callback.beforeValidate(connection);
+        }
+
+        @Override
+        public void afterValidate(Connection connection) {
+            callback.afterValidate(connection);
+        }
+
+        @Override
+        public void beforeBaseline(Connection connection) {
+            callback.beforeBaseline(connection);
+        }
+
+        @Override
+        public void afterBaseline(Connection connection) {
+            callback.afterBaseline(connection);
+        }
+
+        @Override
+        public void beforeRepair(Connection connection) {
+            callback.beforeRepair(connection);
+        }
+
+        @Override
+        public void afterRepair(Connection connection) {
+            callback.afterRepair(connection);
+        }
+
+        @Override
+        public void beforeInfo(Connection connection) {
+            callback.beforeInfo(connection);
+        }
+
+        @Override
+        public void afterInfo(Connection connection) {
+            callback.afterInfo(connection);
+        }
+    }
 }