Spring tips

AccessTokenRequestInterceptor.java

import AuthResult;
import feign.RequestInterceptor;
import feign.RequestTemplate;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.http.HttpHeaders;
import org.springframework.stereotype.Component;

import static org.springframework.util.StringUtils.isEmpty;

@Component
@Slf4j
class AccessTokenRequestInterceptor implements RequestInterceptor {

    private String authorizationHeader;

    @Override
    public void apply(RequestTemplate template) {
        if (isEmpty(authorizationHeader)) {
            return;
        }

        log.debug("Setting Authorization header on HTTP request {} {}", template.method(), template.url());
        template.header(HttpHeaders.AUTHORIZATION, authorizationHeader);
    }

    @EventListener
    public void onAuthFinished(AuthResult authResult) {
        log.debug("Updating access token");
        authorizationHeader = authResult.asAuthorizationHeader();
    }
}

Application.java

import AuthRequest;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.openfeign.EnableFeignClients;

@SpringBootApplication
@EnableConfigurationProperties(AuthRequest.class)
@EnableFeignClients
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

}

application.yaml

server:
  port: 8080

security:
  username: ${HTTPUSER:test}
  password: ${HTTPPASSWORD:pass}
proxy:
  enabled: ${HTTP_PROXY_ENABLED:false}
  settings: ${HTTP_PROXY}

feign:
  httpclient:
    enabled: true
  client:
    config:
      default:
        loggerLevel: ${FEIGN_LOGGER_LEVEL:basic}

spring:
  application:
    name: the-app
  kafka:
    bootstrap-servers: ${KAFKA_CONNECTION:localhost:9093}
    consumer:
      auto-offset-reset: latest
      group-id: the-app-group-id
      enable-auto-commit: false
      max-poll-records: 50
    listener:
      concurrency: 1
      idle-event-interval: 30s
      ack-mode: record
    jaas:
      enabled: true
      login-module: org.apache.kafka.common.security.plain.PlainLoginModule
    properties:
      security.protocol: ${KAFKA_SECURITY_PROTOCOL:SASL_PLAINTEXT}
      sasl.mechanism: PLAIN
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="${KAFKA_SASL_USERNAME:admin}" password="${KAFKA_SASL_PASSWORD:admin-secret}";
  sleuth:
    feign.enabled: false
    #below property is needed because we have 'feign-httpclient' in pom.xml
    web.client.enabled: false
#healthcheck
management:
  endpoints:
    web:
      base-path: /the-app
      exposure:
        include: health,prometheus
  server:
    port: 7070

AuthResult.java

import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;

@Data
public class AuthResult {

    @JsonProperty("access_token")
    private String accessToken;

    @JsonProperty("instance_url")
    private String instanceUrl;

    @JsonProperty("id")
    private String id;

    @JsonProperty("token_type")
    private String tokenType;

    @JsonProperty("issued_at")
    private String issuedAt;

    @JsonProperty("signature")
    private String signature;

    public String asAuthorizationHeader() {
        return tokenType + " " + accessToken;
    }
}

create-kafka-topics.sh

#!/bin/bash

KAFKA_CREATE_TOPICS=$(echo $KAFKA_CREATE_TOPICS | sed 's/,/ /g')

for topic in $KAFKA_CREATE_TOPICS; do
  echo "Creating topic $topic"
  TOPIC_NAME=$(echo $topic | cut -d : -f1)
  TOPIC_PARTITION=$(echo $topic | cut -d : -f2)
  TOPIC_REPL_FACTOR=$(echo $topic | cut -d : -f3)
  kafka-topics --delete --if-exists --topic $TOPIC_NAME --zookeeper $KAFKA_ZOOKEEPER_CONNECT
  kafka-topics --create --topic $TOPIC_NAME --partitions $TOPIC_PARTITION --replication-factor $TOPIC_REPL_FACTOR --if-not-exists --zookeeper $KAFKA_ZOOKEEPER_CONNECT
done

echo "Topics created"

docker-compose.yaml

version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:4.1.1
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    ports:
      - "2181:2181"
  kafka:
    image: confluentinc/cp-kafka:4.1.1
    ports:
      - "9092:9092"
      - "9093:9093"
    depends_on:
      - zookeeper
    environment:
      KAFKA_OPTS: -Djava.security.auth.login.config=/var/docker/config/kafka/kafka_server_jaas.conf
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://kafka:9093,PLAINTEXT://kafka:9092
      KAFKA_LISTENERS: SASL_PLAINTEXT://kafka:9093,PLAINTEXT://kafka:9092
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
      ZOOKEEPER_SASL_ENABLED: 'FALSE'
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - ./.docker/:/var/docker
  schema-registry:
    image: confluentinc/cp-schema-registry:4.1.1
    ports:
      - "8081:8081"
    depends_on:
      - kafka
    environment:
      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper:2181
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - .:/repo
  kafka-create-topics:
    image: confluentinc/cp-kafka:4.1.1
    depends_on:
      - kafka
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CREATE_TOPICS: "topic1:1:1,topic2:1:1"
    command: ["bash", "/var/docker/scripts/wait-for-it.sh", "kafka:9092", "--", "bash", "/var/docker/scripts/create-kafka-topics.sh"]
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - ./.docker/:/var/docker
  kafka-rest:
    image: confluentinc/cp-kafka-rest:4.1.1
    ports:
      - "8082:8082"
    depends_on:
      - kafka
    environment:
      KAFKA_REST_HOST_NAME: kafka-rest
      KAFKA_REST_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
  kafka-topics-ui:
    image: landoop/kafka-topics-ui
    ports:
      - 8000:8000
    depends_on:
      - kafka-rest
    environment:
      KAFKA_REST_PROXY_URL: "http://kafka-rest:8082"
      PROXY: "true"
  wiremock-server:
    image: adoptopenjdk/openjdk11
    ports:
      - "9080:9080"
    volumes:
      - ./wiremock/:/mappings/
    command: sh -c "curl -LO https://search.maven.org/remotecontent?filepath=com/github/tomakehurst/wiremock-standalone/2.22.0/wiremock-standalone-2.22.0.jar && java -jar wiremock-standalone-2.22.0.jar --port 9080 --verbose"

FeignConfig.java

import HttpProxyConfig;
import HttpProxySettings;
import feign.Logger;
import feign.form.FormEncoder;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.impl.client.HttpClientBuilder;
import org.springframework.beans.factory.ObjectFactory;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.http.HttpMessageConverters;
import org.springframework.cloud.openfeign.support.SpringEncoder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@AutoConfigureAfter(HttpProxyConfig.class)
@Slf4j
public class FeignConfig {

    @Bean
    FormEncoder feignFormEncoder(ObjectFactory<HttpMessageConverters> messageConverters) {
        return new FormEncoder(new SpringEncoder(messageConverters));
    }

    @Bean
    @ConditionalOnBean(HttpProxySettings.class)
    HttpClientBuilder httpClientBuilder(HttpProxySettings httpProxySettings) {
        log.debug("Configuring proxy for Feign clients: {}", httpProxySettings);

        return HttpClientBuilder.create()
                .setProxy(httpProxySettings.getHttpHost())
                .setDefaultCredentialsProvider(httpProxySettings.getCredentialsProvider());
    }

    @Bean
    Logger.Level feignLoggerLevel() {
        return Logger.Level.FULL;
    }
}

HttpProxyConfig.java

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConditionalOnProperty("proxy.enabled")
@Slf4j
public class HttpProxyConfig {

    @Bean
    HttpProxySettings httpProxySettings(@Value("${proxy.settings}") String proxySettings) {
        return HttpProxySettings.fromString(proxySettings);
    }
}

HttpProxySettings.java

import lombok.Value;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Value
@Slf4j
public class HttpProxySettings {

    private static final Pattern PROXY_SETTINGS_PATTERN =
            Pattern.compile("^https?://(.+):(.+)@(.+):(\\d+)/?$");
    //                        user (1) -'    |    |      |
    //                             pass (2) -'    |      |
    //                                  host (3) -'      |
    //                                         port (4) -'

    private final String host;
    private final int port;
    private final String user;
    private final String pass;

    public static HttpProxySettings fromString(String input) {
        log.debug("Parsing configuration string: {}", input);

        Matcher matcher = PROXY_SETTINGS_PATTERN.matcher(input);
        if (!matcher.matches()) {
            throw new IllegalArgumentException("Invalid configuration string - expecting 'http(s)://user:pass@host:port'");
        }

        String host = matcher.group(3);
        int port = Integer.parseInt(matcher.group(4));
        String user = matcher.group(1);
        String pass = matcher.group(2);

        return new HttpProxySettings(host, port, user, pass);
    }

    public HttpHost getHttpHost() {
        return new HttpHost(host, port);
    }

    public CredentialsProvider getCredentialsProvider() {
        CredentialsProvider provider = new BasicCredentialsProvider();
        provider.setCredentials(
                new AuthScope(host, port, AuthScope.ANY_REALM, "basic"),
                new UsernamePasswordCredentials(user, pass));
        return provider;
    }
}

kafka_server_jaas.conf

KafkaServer {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="admin"
   password="admin-secret"
   user_admin="admin-secret"
   user_kafkabroker1="kafkabroker1-secret";
};

KafkaConfig.java

import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory) {

        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory);
        factory.getContainerProperties().setAckOnError(false);
        factory.setErrorHandler(new SeekToCurrentErrorHandler());
        return factory;
    }
}

MongoConfig.java

import org.springframework.context.annotation.Bean;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.mongodb.core.convert.MongoCustomConversions;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.List;

public class MongoConfig {

    @Bean
    public MongoCustomConversions mongoCustomConversions() {
        final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");

        return new MongoCustomConversions(List.of(
                (Converter<LocalDate, String>) source -> source.format(dateTimeFormatter),
                (Converter<String, LocalDate>) source -> LocalDate.parse(source, dateTimeFormatter)
        ));
    }

}

SecurityConfig.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.actuate.autoconfigure.security.servlet.EndpointRequest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.authentication.builders.AuthenticationManagerBuilder;
import org.springframework.security.config.annotation.web.builders.HttpSecurity;
import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter;
import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
import org.springframework.security.crypto.password.PasswordEncoder;
import org.springframework.security.web.AuthenticationEntryPoint;
import org.springframework.security.web.authentication.www.BasicAuthenticationEntryPoint;

import static org.springframework.security.config.http.SessionCreationPolicy.STATELESS;

@Configuration
@EnableWebSecurity
public class SecurityConfig extends WebSecurityConfigurerAdapter {

    @Override
    protected void configure(HttpSecurity http) throws Exception {
        http.csrf().disable()
                .sessionManagement().sessionCreationPolicy(STATELESS)
                .and()
                .authorizeRequests()
                .requestMatchers(EndpointRequest.toAnyEndpoint()).permitAll()
                .anyRequest().authenticated()
                .and()
                .httpBasic()
                .authenticationEntryPoint(authenticationEntryPoint());
    }

    @Autowired
    public void configureGlobal(AuthenticationManagerBuilder auth,
                                @Value("${security.username}") String username,
                                @Value("${security.password}") String password) throws Exception {
        auth.inMemoryAuthentication()
                .withUser(username)
                .password(passwordEncoder().encode(password))
                .authorities("ROLE_ALL");
    }

    @Bean
    public AuthenticationEntryPoint authenticationEntryPoint() {
        BasicAuthenticationEntryPoint entryPoint = new BasicAuthenticationEntryPoint();
        entryPoint.setRealmName("auth");
        return entryPoint;
    }

    @Bean
    public PasswordEncoder passwordEncoder() {
        return new BCryptPasswordEncoder();
    }
}

SwaggerConfig.java

import io.swagger.annotations.Api;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.web.bind.annotation.RequestMethod;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.builders.RequestHandlerSelectors;
import springfox.documentation.builders.ResponseMessageBuilder;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;

import java.util.List;

@Configuration
@EnableSwagger2
@Profile("!test")
public class SwaggerConfig {

    @Bean
    public Docket swagger() {
        return new Docket(DocumentationType.SWAGGER_2)
                .select()
                .apis(RequestHandlerSelectors.withClassAnnotation(Api.class))
                .build()
                .useDefaultResponseMessages(false)
                .globalResponseMessage(RequestMethod.POST, List.of(
                        new ResponseMessageBuilder().code(401).message("Unauthorized").build()
                ))
                .apiInfo(apiInfo());
    }

    private ApiInfo apiInfo() {
        return new ApiInfoBuilder()
                .title("Some Service")
                .description("Bla bla bla")
                .version("1.0.0")
                .build();
    }

}

TopicConfig.java

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

@Data
@Configuration
@ConfigurationProperties(prefix = "kafka.topics")
public class TopicConfig {

    private String publisher;

    private String listener;

}

wait-for-it.sh

#!/usr/bin/env bash
#   Use this script to test if a given TCP host/port are available

cmdname=$(basename $0)

echoerr() { if [[ $QUIET -ne 1 ]]; then echo "$@" 1>&2; fi }

usage()
{
    cat << USAGE >&2
Usage:
    $cmdname host:port [-s] [-t timeout] [-- command args]
    -h HOST | --host=HOST       Host or IP under test
    -p PORT | --port=PORT       TCP port under test
                                Alternatively, you specify the host and port as host:port
    -s | --strict               Only execute subcommand if the test succeeds
    -q | --quiet                Don't output any status messages
    -t TIMEOUT | --timeout=TIMEOUT
                                Timeout in seconds, zero for no timeout
    -- COMMAND ARGS             Execute command with args after the test finishes
USAGE
    exit 1
}

wait_for()
{
    if [[ $TIMEOUT -gt 0 ]]; then
        echoerr "$cmdname: waiting $TIMEOUT seconds for $HOST:$PORT"
    else
        echoerr "$cmdname: waiting for $HOST:$PORT without a timeout"
    fi
    start_ts=$(date +%s)
    while :
    do
        if [[ $ISBUSY -eq 1 ]]; then
            nc -z $HOST $PORT
            result=$?
        else
            (echo > /dev/tcp/$HOST/$PORT) >/dev/null 2>&1
            result=$?
        fi
        if [[ $result -eq 0 ]]; then
            end_ts=$(date +%s)
            echoerr "$cmdname: $HOST:$PORT is available after $((end_ts - start_ts)) seconds"
            break
        fi
        sleep 1
    done
    return $result
}

wait_for_wrapper()
{
    # In order to support SIGINT during timeout: http://unix.stackexchange.com/a/57692
    if [[ $QUIET -eq 1 ]]; then
        timeout $BUSYTIMEFLAG $TIMEOUT $0 --quiet --child --host=$HOST --port=$PORT --timeout=$TIMEOUT &
    else
        timeout $BUSYTIMEFLAG $TIMEOUT $0 --child --host=$HOST --port=$PORT --timeout=$TIMEOUT &
    fi
    PID=$!
    trap "kill -INT -$PID" INT
    wait $PID
    RESULT=$?
    if [[ $RESULT -ne 0 ]]; then
        echoerr "$cmdname: timeout occurred after waiting $TIMEOUT seconds for $HOST:$PORT"
    fi
    return $RESULT
}

# process arguments
while [[ $# -gt 0 ]]
do
    case "$1" in
        *:* )
        hostport=(${1//:/ })
        HOST=${hostport[0]}
        PORT=${hostport[1]}
        shift 1
        ;;
        --child)
        CHILD=1
        shift 1
        ;;
        -q | --quiet)
        QUIET=1
        shift 1
        ;;
        -s | --strict)
        STRICT=1
        shift 1
        ;;
        -h)
        HOST="$2"
        if [[ $HOST == "" ]]; then break; fi
        shift 2
        ;;
        --host=*)
        HOST="${1#*=}"
        shift 1
        ;;
        -p)
        PORT="$2"
        if [[ $PORT == "" ]]; then break; fi
        shift 2
        ;;
        --port=*)
        PORT="${1#*=}"
        shift 1
        ;;
        -t)
        TIMEOUT="$2"
        if [[ $TIMEOUT == "" ]]; then break; fi
        shift 2
        ;;
        --timeout=*)
        TIMEOUT="${1#*=}"
        shift 1
        ;;
        --)
        shift
        CLI=("$@")
        break
        ;;
        --help)
        usage
        ;;
        *)
        echoerr "Unknown argument: $1"
        usage
        ;;
    esac
done

if [[ "$HOST" == "" || "$PORT" == "" ]]; then
    echoerr "Error: you need to provide a host and port to test."
    usage
fi

TIMEOUT=${TIMEOUT:-15}
STRICT=${STRICT:-0}
CHILD=${CHILD:-0}
QUIET=${QUIET:-0}

# check to see if timeout is from busybox?
# check to see if timeout is from busybox?
TIMEOUT_PATH=$(realpath $(which timeout))
if [[ $TIMEOUT_PATH =~ "busybox" ]]; then
        ISBUSY=1
        BUSYTIMEFLAG="-t"
else
        ISBUSY=0
        BUSYTIMEFLAG=""
fi

if [[ $CHILD -gt 0 ]]; then
    wait_for
    RESULT=$?
    exit $RESULT
else
    if [[ $TIMEOUT -gt 0 ]]; then
        wait_for_wrapper
        RESULT=$?
    else
        wait_for
        RESULT=$?
    fi
fi

if [[ $CLI != "" ]]; then
    if [[ $RESULT -ne 0 && $STRICT -eq 1 ]]; then
        echoerr "$cmdname: strict mode, refusing to execute subprocess"
        exit $RESULT
    fi
    exec "${CLI[@]}"
else
    exit $RESULT
fi