public interface EventLogWriter {

    /**
     * Fires a data change event about a <b>creation</b> of some resource
     * (object).
     *
     * @param eventType
     *            the Nakadi event type of the event. This is roughly equivalent
     *            to an event channel or topic.
     *
     * @param dataType
     *            the content of the {@code data_type} field of the Nakadi
     *            event.
     *
     * @param data
     *            some POJO that can be serialized into JSON (required
     *            parameter). This is meant to be a representation of the
     *            resource which was created. It will be used as content of the
     *            {@code data} field of the Nakadi event.
     */
    @Transactional
    void fireCreateEvent(String eventType, String dataType, Object data);

    /**
     * Fires a data change event about an update of some resource (object).
     *
     * @param eventType
     *            the Nakadi event type of the event. This is roughly equivalent
     *            to an event channel or topic.
     *
     * @param dataType
     *            the content of the {@code data_type} field of the Nakadi
     *            event.
     *
     * @param data
     *            some POJO that can be serialized into JSON (required
     *            parameter). This is meant to be a representation of the new
     *            state of the resource which was updated. It will be used as
     *            content of the {@code data} field of the Nakadi event.
     */
    @Transactional
    void fireUpdateEvent(String eventType, String dataType, Object data);

    /**
     * Fires a data change event about the deletion of some resource (object).
     *
     * @param eventType
     *            the Nakadi event type of the event. This is roughly equivalent
     *            to an event channel or topic.
     *
     * @param dataType
     *            the content of the {@code data_type} field of the Nakadi
     *            event.
     *
     * @param data
     *            some POJO that can be serialized into JSON (required
     *            parameter). This is meant to be a representation of the last
     *            state (before the deletion) of the resource which was deleted.
     *            It will be used as content of the {@code data} field of the
     *            Nakadi event.
     */
    @Transactional
    void fireDeleteEvent(String eventType, String dataType, Object data);

    /**
     * Fires a data change event with a snapshot of some resource (object).
     * <p>
     * This notifies your consumers about the current state of a resource, even
     * if nothing changed. Typical use cases include initial replication to new
     * consumers or hotfixes of data inconsistencies between producer and
     * consumer.
     * </p>
     * <p>
     * Normally applications don't have to call this themselves, instead they
     * should implement the
     * {@link SnapshotEventGenerator}
     * interface to add support for snapshot creation via the actuator endpoint.
     * </p>
     *
     * @param eventType
     *            the Nakadi event type of the event. This is roughly equivalent
     *            to an event channel or topic.
     *
     * @param dataType
     *            the content of the {@code data_type} field of the Nakadi
     *            event.
     *
     * @param data
     *            some POJO that can be serialized into JSON (required
     *            parameter). This is meant to be a representation of the
     *            current state of the resource. It will be used as content of
     *            the {@code data} field of the Nakadi event.
     */
    @Transactional
    void fireSnapshotEvent(String eventType, String dataType, Object data);

    /**
     * Fires a business event, i.e. an event communicating the fact that some
     * business process step happened. The payload object will be used as the
     * main event content (just metadata will be added). Same as for data change
     * events, you should call this method in the same transaction as you are
     * storing related changes into your database.
     *
     * @param eventType
     *            the Nakadi event type of the event. This is roughly equivalent
     *            to an event channel or topic.
     *
     * @param payload
     *            some POJO that can be serialized into JSON (required
     *            parameter)
     */
    @Transactional
    void fireBusinessEvent(String eventType, Object payload);
} public interface EventLogRepository {
    Collection<EventLog> findByLockedByAndLockedUntilGreaterThan(String lockedBy, Instant lockedUntil);

    void lockSomeMessages(String lockId, Instant now, Instant lockExpires);

    void delete(EventLog eventLog);

    void persist(EventLog eventLog);

    void deleteAll();

    EventLog findOne(Integer id);
} public class EventLogRepositoryImpl implements EventLogRepository {
    private NamedParameterJdbcTemplate jdbcTemplate;

    public EventLogRepositoryImpl(NamedParameterJdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @Override
    public Collection<EventLog> findByLockedByAndLockedUntilGreaterThan(String lockedBy, Instant lockedUntil) {
        Map<String, Object> namedParameterMap = new HashMap<>();
        namedParameterMap.put("lockedBy", lockedBy);
        namedParameterMap.put("lockedUntil", toSqlTimestamp(lockedUntil));
        return jdbcTemplate.query(
            "SELECT * FROM nakadi_events.event_log where locked_by = :lockedBy and locked_until > :lockedUntil",
            namedParameterMap,
            new BeanPropertyRowMapper<>(EventLog.class)
        );
    }

    @Override
    public void lockSomeMessages(String lockId, Instant now, Instant lockExpires) {
        Map<String, Object> namedParameterMap = new HashMap<>();
        namedParameterMap.put("lockId", lockId);
        namedParameterMap.put("now", toSqlTimestamp(now));
        namedParameterMap.put("lockExpires", toSqlTimestamp(lockExpires));
        jdbcTemplate.update(
            "UPDATE nakadi_events.event_log SET locked_by = :lockId, locked_until = :lockExpires where locked_until is null or locked_until < :now",
            namedParameterMap
        );
    }

    @Override
    public void delete(EventLog eventLog) {
        Map<String, Object> namedParameterMap = new HashMap<>();
        namedParameterMap.put("id", eventLog.getId());
        jdbcTemplate.update(
            "DELETE FROM nakadi_events.event_log where id = :id",
            namedParameterMap
        );
    }

    @Override
    public void persist(EventLog eventLog) {
        Timestamp now = toSqlTimestamp(Instant.now());
        MapSqlParameterSource namedParameterMap = new MapSqlParameterSource();
        namedParameterMap.addValue("eventType", eventLog.getEventType());
        namedParameterMap.addValue("eventBodyData", eventLog.getEventBodyData());
        namedParameterMap.addValue("flowId", eventLog.getFlowId());
        namedParameterMap.addValue("created", now);
        namedParameterMap.addValue("lastModified", now);
        namedParameterMap.addValue("lockedBy", eventLog.getLockedBy());
        namedParameterMap.addValue("lockedUntil", eventLog.getLockedUntil());
        GeneratedKeyHolder generatedKeyHolder = new GeneratedKeyHolder();
        jdbcTemplate.update(
            "INSERT INTO " +
                "    nakadi_events.event_log " +
                "    (event_type, event_body_data, flow_id, created, last_modified, locked_by, locked_until) " +
                "VALUES " +
                "    (:eventType, :eventBodyData, :flowId, :created, :lastModified, :lockedBy, :lockedUntil)",
            namedParameterMap,
            generatedKeyHolder
        );

        eventLog.setId((Integer) generatedKeyHolder.getKeys().get("id"));
    }

    private Timestamp toSqlTimestamp(Instant now) {
        if (now == null) {
            return null;
        }
        return Timestamp.from(now);
    }

    @Override
    public void deleteAll() {
        jdbcTemplate.update("DELETE from nakadi_events.event_log", new HashMap<>());
    }

    @Override
    public EventLog findOne(Integer id) {
        Map<String, Object> namedParameterMap = new HashMap<>();
        namedParameterMap.put("id", id);
        try {
            return jdbcTemplate.queryForObject(
                "SELECT * FROM nakadi_events.event_log where id = :id",
                namedParameterMap,
                new BeanPropertyRowMapper<>(EventLog.class)
            );
        } catch (EmptyResultDataAccessException ignored) {
            return null;
        }
    }
} public class EventLogWriterImpl implements EventLogWriter {

    private final EventLogRepository eventLogRepository;

    private final ObjectMapper objectMapper;
    private final FlowIdComponent flowIdComponent;

    public EventLogWriterImpl(EventLogRepository eventLogRepository, ObjectMapper objectMapper, FlowIdComponent flowIdComponent) {
        this.eventLogRepository = eventLogRepository;
        this.objectMapper = objectMapper;
        this.flowIdComponent = flowIdComponent;
    }

    @Override
    @Transactional
    public void fireCreateEvent(final String eventType, final String dataType, final Object data) {
        final EventLog eventLog = createEventLog(eventType, new DataChangeEventEnvelope(CREATE.toString(), dataType, data));
        eventLogRepository.persist(eventLog);
    }

    @Override
    @Transactional
    public void fireUpdateEvent(final String eventType, final String dataType, final Object data) {
        final EventLog eventLog = createEventLog(eventType, new DataChangeEventEnvelope(UPDATE.toString(), dataType, data));
        eventLogRepository.persist(eventLog);
    }

    @Override
    @Transactional
    public void fireDeleteEvent(final String eventType, final String dataType, final Object data) {
        final EventLog eventLog = createEventLog(eventType, new DataChangeEventEnvelope(DELETE.toString(), dataType, data));
        eventLogRepository.persist(eventLog);
    }

    @Override
    @Transactional
    public void fireSnapshotEvent(final String eventType, final String dataType, final Object data) {
        final EventLog eventLog = createEventLog(eventType, new DataChangeEventEnvelope(SNAPSHOT.toString(), dataType, data));
        eventLogRepository.persist(eventLog);
    }

    @Override
    @Transactional
    public void fireBusinessEvent(final String eventType, Object payload) {
        final EventLog eventLog = createEventLog(eventType, payload);
        eventLogRepository.persist(eventLog);
    }

    private EventLog createEventLog(final String eventType, final Object eventPayload) {
        final EventLog eventLog = new EventLog();
        eventLog.setEventType(eventType);
        try {
            eventLog.setEventBodyData(objectMapper.writeValueAsString(eventPayload));
        } catch (final JsonProcessingException e) {
            throw new IllegalStateException("could not map object to json: " + eventPayload.toString(), e);
        }

        eventLog.setFlowId(flowIdComponent.getXFlowIdValue());
        return eventLog;
    }

} public interface FlowIdComponent {
    String getXFlowIdValue();

    void startTraceIfNoneExists();
} @Slf4j
public class NoopFlowIdComponent implements FlowIdComponent {

    @Override
    public String getXFlowIdValue() {
        return null;
    }

    @Override
    public void startTraceIfNoneExists() {
    }
} @Slf4j
public class TracerFlowIdComponent implements FlowIdComponent {
    
    private static final String X_FLOW_ID = "X-Flow-ID";
    
    private final Tracer tracer;
    
    public TracerFlowIdComponent(Tracer tracer) {
        this.tracer = tracer;
    }

    public String getXFlowIdKey() {
        return X_FLOW_ID;
    }

    @Override
    public String getXFlowIdValue() {
        if (tracer != null) {
            try {
                return tracer.get(X_FLOW_ID).getValue();
            } catch (IllegalArgumentException e) {
                log.warn("No trace was configured for the name {}. Returning null. " +
                        "To configure Tracer provide an application property: " +
                        "tracer.traces.X-Flow-ID=flow-id", X_FLOW_ID);
            } catch (IllegalStateException e) {
                log.warn("Unexpected Error while receiving the Trace Id {}. Returning null. " +
                    "Please check your tracer configuration: {}", X_FLOW_ID, e.getMessage());
            }
        } else {
            log.warn("No bean of class Tracer was found. Returning null.");
        }
        return null;
    }

    @Override
    public void startTraceIfNoneExists() {
        if (tracer != null) {
            try {
                tracer.get(X_FLOW_ID).getValue();
            } catch (IllegalArgumentException e) {
                log.warn("No trace was configured for the name {}. Returning null. " +
                        "To configure Tracer provide an application property: " +
                        "tracer.traces.X-Flow-ID=flow-id", X_FLOW_ID);
            } catch (IllegalStateException e) {
                tracer.start();
            }
        } else {
            log.warn("No bean of class Tracer was found.");
        }
    }
} public interface SnapshotEventGenerator {

    /**
     * <p>
     * Returns a batch of snapshots of given type (event type is an event
     * channel topic name). The implementation may return an arbitrary amount of
     * results, but it must return at least one element if there are entities
     * matching the parameters.
     * </p>
     * <p>
     * Calling this method must have no side effects.
     * </p>
     * The library will call your implementation like this:
     * <ul>
     * <li>Request: generateSnapshots(null, filter), Response: 1,2,3</li>
     * <li>Request: generateSnapshots(3, filter), Response: 4,5</li>
     * <li>Request: generateSnapshots(5, filter), Response: emptyList</li>
     * </ul>
     * <p>
     * It is your responsibility to make sure that the returned events are
     * ordered by their ID ascending and that, given you return a list of events
     * for entities with IDs {id<sub>1</sub>, ..., id<sub>N</sub>}, there exists
     * no entity with an ID between id<sub>1</sub> and id<sub>N</sub>, that is
     * not part of the result.
     * </p>
     *
     * @param withIdGreaterThan
     *            if not null, only events for entities with an ID greater than
     *            the given one must be returned
     *
     * @param filter
     *            a filter for the snapshot generation mechanism. This value is
     *            simply passed through from the request body of the REST
     *            endpoint (or from any other triggering mechanism). If there
     *            was no request body, this will be {@code null}.
     *
     *            Implementors can interpret it in whatever way they want (even
     *            ignore it). All calls for one snapshot generation will receive
     *            the same string.
     *
     * @return list of elements (wrapped in Snapshot objects) ordered by their
     *         ID.
     */
    List<Snapshot> generateSnapshots(Object withIdGreaterThan, String filter);

    /**
     * The name of the event type supported by this snapshot generator.
     */
    String getSupportedEventType();
} public class SnapshotCreationService {

    private final Map<String, SnapshotEventGenerator> snapshotEventProviders;

    private final EventLogWriter eventLogWriter;

    /**
     * Creates the service.
     *
     * @param snapshotEventGenerators
     *            the event generators. Each of them must have a different
     *            supported event type.
     * @param eventLogWriter
     *            The event log writer to which the newly generated snapshot
     *            events are pushed.
     *            @throws IllegalStateException if two event generators declare to be responsible for the same event type.
     */
    public SnapshotCreationService(List<SnapshotEventGenerator> snapshotEventGenerators,
            EventLogWriter eventLogWriter) {
        this.snapshotEventProviders = snapshotEventGenerators.stream()
                .collect(toMap(SnapshotEventGenerator::getSupportedEventType, identity()));
        this.eventLogWriter = eventLogWriter;
    }

    public void createSnapshotEvents(final String eventType, String filter) {
        final SnapshotEventGenerator snapshotEventGenerator = snapshotEventProviders.get(eventType);
        if (snapshotEventGenerator == null) {
            throw new UnknownEventTypeException(eventType);
        }

        Object lastProcessedId = null;
        do {
            final List<Snapshot> snapshots = snapshotEventGenerator.generateSnapshots(lastProcessedId, filter);
            if (snapshots.isEmpty()) {
                break;
            }

            for (final Snapshot snapshot : snapshots) {
                eventLogWriter.fireSnapshotEvent(eventType, snapshot.getDataType(), snapshot.getData());
                lastProcessedId = snapshot.getId();
            }
        } while (true);
    }

    public Set<String> getSupportedEventTypes() {
        return unmodifiableSet(snapshotEventProviders.keySet());
    }
} @Endpoint(id = "snapshot-event-creation")
public class SnapshotEventCreationEndpoint {
    private final SnapshotCreationService snapshotCreationService;
    private final FlowIdComponent flowIdComponent;

    public SnapshotEventCreationEndpoint(SnapshotCreationService snapshotCreationService, FlowIdComponent flowIdComponent) {
        this.snapshotCreationService = snapshotCreationService;
        this.flowIdComponent = flowIdComponent;
    }

    @ReadOperation
    public SnapshotReport getSupportedEventTypes() {
        return new SnapshotReport(snapshotCreationService.getSupportedEventTypes());
    }

    @WriteOperation
    public void createFilteredSnapshotEvents(
            // this is the event type. Could have a better name, but since Spring Boot relies on the -parameters
            // compiler flag being set to resolve path parameter names, it would then get trickier to reliably run this
            // Test in the IDE. So let's stick with arg0 for now.
            @Selector String arg0,
            @Nullable String filter) {
        flowIdComponent.startTraceIfNoneExists();
        snapshotCreationService.createSnapshotEvents(arg0, filter);
    }


    @AllArgsConstructor
    @Getter
    public static class SnapshotReport {
        private final Set<String> supportedEventTypes;
    }

} public interface NakadiPublishingClient {
    void publish(String eventType, List<?> nakadiEvents) throws Exception;
} @Slf4j
public class EventTransmissionService {

    private final EventLogRepository eventLogRepository;
    private final NakadiPublishingClient nakadiPublishingClient;
    private final ObjectMapper objectMapper;
    private final int lockDuration;
    private final int lockDurationBuffer;

    private Clock clock = Clock.systemDefaultZone();

    public EventTransmissionService(EventLogRepository eventLogRepository, NakadiPublishingClient nakadiPublishingClient, ObjectMapper objectMapper,
    int lockDuration, int lockDurationBuffer) {
        this.eventLogRepository = eventLogRepository;
        this.nakadiPublishingClient = nakadiPublishingClient;
        this.objectMapper = objectMapper;
        this.lockDuration = lockDuration;
        this.lockDurationBuffer = lockDurationBuffer;
    }

    @Transactional
    public Collection<EventLog> lockSomeEvents() {
        String lockId = UUID.randomUUID().toString();
        log.debug("Locking events for replication with lockId {} for {} seconds", lockId, lockDuration);
        eventLogRepository.lockSomeMessages(lockId, now(), now().plus(lockDuration, SECONDS));
        return eventLogRepository.findByLockedByAndLockedUntilGreaterThan(lockId, now());
    }

    @Transactional
    public void sendEvents(Collection<EventLog> events) {
        EventBatcher batcher = new EventBatcher(objectMapper, this::publishBatch);

        for (EventLog event : events) {
            if (lockNearlyExpired(event)) {
                // to avoid that two instances process this event, we skip it
                continue;
            }

            NakadiEvent nakadiEvent;

            try {
                nakadiEvent = mapToNakadiEvent(event);
            } catch (Exception e) {
                log.error("Could not serialize event {} of type {}, skipping it.", event.getId(), event.getEventType(), e);
                continue;
            }

            batcher.pushEvent(event, nakadiEvent);
        }

        batcher.finish();
    }

    /**
     * Publishes a list of events.
     * All of the events in this list need to be destined for the same event type.
     */
    private void publishBatch(List<BatchItem> batch) {
        try {
            this.tryToPublishBatch(batch);
        } catch (Exception e) {
            log.error("Could not send {} events of type {}, skipping them.", batch.size(), batch.get(0).getEventLogEntry().getEventType(), e);
        }
    }

    /**
     * Tries to publish a set of events (all of which need to belong to the same event type).
     * The successful ones will be deleted from the database.
     */
    private void tryToPublishBatch(List<BatchItem> batch) throws Exception {
        Stream<EventLog> successfulEvents;
        String eventType = batch.get(0).getEventLogEntry().getEventType();
        try {
            nakadiPublishingClient.publish(
                    eventType,
                    batch.stream()
                            .map(BatchItem::getNakadiEvent)
                            .collect(Collectors.toList())
            );
            successfulEvents = batch.stream().map(BatchItem::getEventLogEntry);
            log.info("Sent {} events of type {}.", batch.size(), eventType);
        } catch (EventPublishingException e) {
            log.error("{} out of {} events of type {} failed to be sent. Exception ",
                    e.getResponses().length, batch.size(), eventType, e);
            List<String> failedEids = collectEids(e);
            successfulEvents =
                    batch.stream()
                            .map(BatchItem::getEventLogEntry)
                            .filter(rawEvent -> !failedEids.contains(convertToUUID(rawEvent.getId())));
        }

        successfulEvents.forEach(eventLogRepository::delete);
    }

    private List<String> collectEids(EventPublishingException e) {
        return Arrays.stream(e.getResponses()).map(BatchItemResponse::getEid).collect(Collectors.toList());
    }

    private boolean lockNearlyExpired(EventLog eventLog) {
        // since clocks never work exactly synchronous and sending the event also takes some time, we include a safety
        // buffer here. This is still not 100% precise, but since we require events to be consumed idempotent, sending
        // one event twice wont hurt much.
        return now().isAfter(eventLog.getLockedUntil().minus(lockDurationBuffer, SECONDS));
    }

    private NakadiEvent mapToNakadiEvent(final EventLog event) throws IOException {
        final NakadiEvent nakadiEvent = new NakadiEvent();

        final NakadiMetadata metadata = new NakadiMetadata();
        metadata.setEid(convertToUUID(event.getId()));
        metadata.setOccuredAt(event.getCreated());
        metadata.setFlowId(event.getFlowId());
        nakadiEvent.setMetadata(metadata);

        LinkedHashMap<String, Object> payloadDTO = objectMapper.readValue(event.getEventBodyData(), new TypeReference<LinkedHashMap<String, Object>>() { });

        nakadiEvent.setData(payloadDTO);

        return nakadiEvent;
    }

    private Instant now() {
        return clock.instant();
    }

    public void overrideClock(Clock clock) {
        this.clock = clock;
    }

    /**
     * Converts a number in UUID format.
     *
     * <p>For instance 213 will be converted to "00000000-0000-0000-0000-0000000000d5"</p>
     */
    private String convertToUUID(final int number) {
        return new UUID(0, number).toString();
    }

} public class EventTransmitter {
    private final EventTransmissionService eventTransmissionService;

    public EventTransmitter(EventTransmissionService eventTransmissionService) {
        this.eventTransmissionService = eventTransmissionService;
    }

    public void sendEvents() {
        eventTransmissionService.sendEvents(eventTransmissionService.lockSomeEvents());
    }
} public class FahrscheinNakadiPublishingClient implements NakadiPublishingClient {
    private final org.zalando.fahrschein.NakadiClient delegate;

    public FahrscheinNakadiPublishingClient(org.zalando.fahrschein.NakadiClient delegate) {
        this.delegate = delegate;
    }

    @Override
    public void publish(String eventType, List<?> nakadiEvents) throws Exception {
        delegate.publish(eventType, nakadiEvents);
    }
}
import java.net.URI;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
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.flyway.FlywayProperties;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.zalando.fahrschein.NakadiClient;
import org.zalando.nakadiproducer.eventlog.EventLogWriter;
import org.zalando.nakadiproducer.eventlog.impl.EventLogRepository;
import org.zalando.nakadiproducer.eventlog.impl.EventLogRepositoryImpl;
import org.zalando.nakadiproducer.eventlog.impl.EventLogWriterImpl;
import org.zalando.nakadiproducer.flowid.FlowIdComponent;
import org.zalando.nakadiproducer.flowid.NoopFlowIdComponent;
import org.zalando.nakadiproducer.flowid.TracerFlowIdComponent;
import org.zalando.nakadiproducer.snapshots.SnapshotEventGenerator;
import org.zalando.nakadiproducer.snapshots.impl.SnapshotCreationService;
import org.zalando.nakadiproducer.snapshots.impl.SnapshotEventCreationEndpoint;
import org.zalando.nakadiproducer.transmission.NakadiPublishingClient;
import org.zalando.nakadiproducer.transmission.impl.EventTransmissionService;
import org.zalando.nakadiproducer.transmission.impl.EventTransmitter;
import org.zalando.nakadiproducer.transmission.impl.FahrscheinNakadiPublishingClient;
import org.zalando.tracer.Tracer;
import com.fasterxml.jackson.databind.ObjectMapper;
package org.zalando.nakadiproducer;




@Configuration
@AutoConfigureAfter(name="org.zalando.tracer.spring.TracerAutoConfiguration")
@EnableScheduling
@EnableConfigurationProperties({ DataSourceProperties.class, FlywayProperties.class })
@Slf4j
public class NakadiProducerAutoConfiguration {

    @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) {
            return new FahrscheinNakadiPublishingClient(NakadiClient.builder(nakadiBaseUri)
                    .withAccessTokenProvider(accessTokenProvider::getAccessToken).build());
        }

        @ConditionalOnClass(name = "org.zalando.stups.tokens.Tokens")
        @Configuration
        static class StupsTokenConfiguration {
            @Bean(destroyMethod = "stop")
            @ConditionalOnMissingBean(AccessTokenProvider.class)
            public StupsTokenComponent accessTokenProvider(
                    @Value("${nakadi-producer.access-token-uri:http://nakadi-producer.access-token-uri.not-set}") URI accessTokenUri,
                    @Value("${nakadi-producer.access-token-scopes:uid}") String[] accessTokenScopes) {
                return new StupsTokenComponent(accessTokenUri, Arrays.asList(accessTokenScopes));
            }
        }
    }

    @ConditionalOnMissingBean(NakadiPublishingClient.class)
    @ConditionalOnBean(NakadiClient.class)
    @Configuration
    static class ExistingFahrscheinNakadiClientConfiguration {

        @Bean
        public NakadiPublishingClient nakadiProducerPublishingClient(NakadiClient fahrscheinNakadiClient) {
            return new FahrscheinNakadiPublishingClient(fahrscheinNakadiClient);
        }
    }

    @Bean
    @ConditionalOnMissingClass("org.zalando.tracer.Tracer")
    @ConditionalOnMissingBean(FlowIdComponent.class)
    public FlowIdComponent flowIdComponent() {
        return new NoopFlowIdComponent();
    }

    @ConditionalOnClass(name = "org.zalando.tracer.Tracer")
    @Configuration
    static class TracerConfiguration {
        @Autowired(required = false)
        private Tracer tracer;

        @SuppressWarnings("SpringJavaAutowiringInspection")
        @ConditionalOnMissingBean(FlowIdComponent.class)
        @Bean
        public FlowIdComponent flowIdComponent() {
            if (tracer == null) {
                return new NoopFlowIdComponent();
            } else {
                return new TracerFlowIdComponent(tracer);
            }
        }
    }

    @Bean
    @ConditionalOnMissingBean
    public SnapshotEventCreationEndpoint snapshotEventCreationEndpoint(
            SnapshotCreationService snapshotCreationService, FlowIdComponent flowIdComponent) {
        return new SnapshotEventCreationEndpoint(snapshotCreationService, flowIdComponent);
    }

    @Bean
    public SnapshotCreationService snapshotCreationService(
            Optional<List<SnapshotEventGenerator>> snapshotEventGenerators,
            EventLogWriter eventLogWriter) {