Merge branch 'master' into release-candidate-21.0.0-RC2

Author: ePaul

README.md

@@ -130,6 +130,21 @@ nakadi-producer:
  access-token-uri: https://token.auth.example.org/oauth2/access_token
 ```
 
+By default, the initialized Fahrschein Nakadi client uses the `SimpleRequestFactory` with `ContentEncoding.GZIP`.
+If you want to override this setting (e.g. to enable zstd compression), you can create an overwriting configuration property:
+```yaml
+nakadi-producer:
+  content-encoding: ZSTD
+```
+If you want to disable compression completely:
+```yaml
+nakadi-producer:
+  content-encoding: IDENTITY
+```
+As we gain experience, the default value might change in future versions of this library, so if you need to stay on gzip compression, use `content-encoding: GZIP`.
+
+Alternatively, you can define your own bean of type `RequestFactory`, which will then be used instead of ours.
+
 #### OAuth (scope) configuration in a non-Zalando environment
 Please consult the [manual of Zalando's tokens library](https://github.com/zalando/tokens) for more configuration options (like `CREDENTIALS_DIR` or via environment variables.
 

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

@@ -23,6 +23,10 @@
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.zalando.fahrschein.NakadiClient;
+import org.zalando.fahrschein.NakadiClientBuilder;
+import org.zalando.fahrschein.http.api.ContentEncoding;
+import org.zalando.fahrschein.http.api.RequestFactory;
+import org.zalando.fahrschein.http.simple.SimpleRequestFactory;
 import org.zalando.nakadiproducer.eventlog.EventLogWriter;
 import org.zalando.nakadiproducer.eventlog.impl.EventLogRepository;
 import org.zalando.nakadiproducer.eventlog.impl.EventLogRepositoryImpl;
@@ -55,8 +59,8 @@ static class FahrscheinWithTokensNakadiClientConfiguration {
 
         @Bean
         public NakadiPublishingClient nakadiProducerPublishingClient(AccessTokenProvider accessTokenProvider,
-                @Value("${nakadi-producer.nakadi-base-uri}") URI nakadiBaseUri) {
-            return new FahrscheinNakadiPublishingClient(NakadiClient.builder(nakadiBaseUri)
+                @Value("${nakadi-producer.nakadi-base-uri}") URI nakadiBaseUri, RequestFactory requestFactory) {
+            return new FahrscheinNakadiPublishingClient(NakadiClient.builder(nakadiBaseUri, requestFactory)
                     .withAccessTokenProvider(accessTokenProvider::getAccessToken).build());
         }
 
@@ -70,7 +74,14 @@ 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){
+            return new SimpleRequestFactory(encoding);
         }
+
     }
 
     @ConditionalOnMissingBean(NakadiPublishingClient.class)
@@ -164,5 +175,4 @@ public EventTransmissionService eventTransmissionService(
     public FlywayMigrator flywayMigrator() {
         return new FlywayMigrator();
     }
-
 }

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

@@ -0,0 +1,38 @@
+package org.zalando.nakadiproducer;
+
+import lombok.SneakyThrows;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.zalando.fahrschein.http.api.ContentEncoding;
+import org.zalando.fahrschein.http.api.RequestFactory;
+import org.zalando.nakadiproducer.config.EmbeddedDataSourceConfig;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+// this class has no @ActiveProfiles("test"), so it doesn't use the MockNakadiClient.
+@SpringBootTest(
+        webEnvironment = SpringBootTest.WebEnvironment.MOCK,
+        properties = {
+                "nakadi-producer.scheduled-transmission-enabled:false",
+                // as we are not defining a mock nakadi client, we need to provide these properties:
+                "nakadi-producer.encoding:ZSTD",
+                "nakadi-producer.nakadi-base-uri:http://nakadi.example.com/",
+        },
+        classes = { TestApplication.class, EmbeddedDataSourceConfig.class }
+)
+public class NakadiClientContentEncodingIT {
+
+    @Autowired
+    private RequestFactory requestFactory;
+
+    @Test
+    @SneakyThrows
+    public void pickUpContentEncodingFromConfig() {
+        final ContentEncoding contentEncoding =
+                (ContentEncoding) FieldUtils.readField(requestFactory, "contentEncoding", true);
+        assertThat(contentEncoding, is(ContentEncoding.ZSTD));
+    }
+}

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

@@ -4,7 +4,6 @@
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
 @SpringBootApplication
-@EnableNakadiProducer
 public class TestApplication {
 
     public static void main(final String[] args) {

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

@@ -2,10 +2,13 @@
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
 import org.zalando.nakadiproducer.transmission.MockNakadiPublishingClient;
 import org.zalando.nakadiproducer.transmission.NakadiPublishingClient;
 
 @Configuration
+// we only want this in the tests which actually want a mock, i.e. ones which have a "test" profile.
+@Profile("test")
 public class MockNakadiClientConfig {
     @Bean
     public NakadiPublishingClient nakadiClient() {

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

@@ -31,7 +31,7 @@
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-jdbc</artifactId>
-            <version>2.5.5</version>
+            <version>2.5.6</version>
         </dependency>
         <dependency>
             <groupId>com.opentable.components</groupId>

nakadi-producer/pom.xml

@@ -44,7 +44,7 @@
         <dependency>
             <groupId>org.zalando</groupId>
             <artifactId>fahrschein</artifactId>
-            <version>0.18.0</version>
+            <version>0.24.0</version>
         </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>

nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/MockNakadiPublishingClient.java

@@ -4,10 +4,8 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.PropertyNamingStrategy;
-import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
-import com.fasterxml.jackson.module.paramnames.ParameterNamesModule;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 
@@ -54,9 +52,9 @@ private String transformToJson(Object value) {
 
     private static ObjectMapper createDefaultObjectMapper() {
         final ObjectMapper objectMapper = new ObjectMapper();
-        objectMapper.setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE);
+        objectMapper.setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
         objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
-        objectMapper.registerModules(new Jdk8Module(), new ParameterNamesModule(), new JavaTimeModule());
+        objectMapper.registerModules(new JavaTimeModule());
         objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
         return objectMapper;
     }