Fahrschein

https://github.com/zalando-nakadi/fahrschein
Admin
Fahrschein is a client library for the Nakadi event bus, providing consistent error handling,...
Tokens:6,287
Snippets:92
Trust Score:6.9
License:Apache-2.0
Update:3 months ago
Show doc for...
Tokens:
Raw
### Start Nakadi with Docker Compose

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-example/README.md

Initializes the Nakadi environment required for running the examples.

```sh
docker-compose up -d fahrschein-example/src/test/resources/docker-compose.yaml
```

--------------------------------

### Initialize Low-level API Client

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Setup a NakadiClient using the InMemoryCursorManager for low-level event streaming.

```java
CursorManager cursorManager = new InMemoryCursorManager();

NakadiClient nakadiClient = NakadiClient.builder(NAKADI_URI, new SimpleRequestFactory(ContentEncoding.IDENTITY))
        .withAccessTokenProvider(new PlatformAccessTokenProvider())
        .withCursorManager(cursorManager)
        .build();

nakadiClient.stream(eventName)
        .listen(SalesOrderPlaced.class, listener);
```

--------------------------------

### Initialize Partition Offsets

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Configure the starting point for event streaming by managing partition offsets via the CursorManager.

```java
List<Partition> partitions = nakadiClient.getPartitions(eventName);

// NakadiClient can be configured to start reading from the oldest available offset in each partition
nakadiClient.stream(eventName).readFromBegin(partitions);

// Or from the newest available offset, but this is the same as the default
nakadiClient.stream(eventName).readFromNewestAvailableOffset(partitions);

// Or (for a persistent cursor manager) we can start reading from the last offset that we processed if it's still available, and from the oldest available offset otherwise
nakadiClient.stream(eventName).skipUnavailableOffsets(partitions);
```

--------------------------------

### Run end-to-end tests with manual Docker management

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Start required Docker containers manually before executing the end-to-end test suite.

```sh
docker-compose -f fahrschein-e2e-test/src/test/resources/docker-compose.yaml up -d
./gradlew :fahrschein-e2e:check -Pe2e.composeProvided
```

--------------------------------

### Manage event stream lifecycle in Java

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Demonstrates how to start, stop, and resume a Nakadi event stream using an ExecutorService and the runnable interface.

```java
ExecutorService executorService = Executors.newSingleThreadExecutor();

Runnable runnable = nakadiClient.stream(SALES_ORDER_SERVICE_ORDER_PLACED)
        .runnable(SalesOrderPlaced.class, listener)
        .unchecked();

// start consuming events
Future<?> future = executorService.submit(runnable);

// stop consuming events
future.cancel(true);

// resume consuming events
Future<?> future2 = executorService.submit(runnable);
```

--------------------------------

### Extract OpenTelemetry span context from event metadata

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-opentelemetry/README.md

Manually extract the parent context from event metadata and start a new consumer span.

```java
Context parentContext = OpenTelemetryHelper.extractFromMetadata(event.getMetadata());
Span span = tracer.spanBuilder("sample-consuming-operation")
    .setParent(eventContext)
    .setSpanKind(SpanKind.CONSUMER)
    .startSpan();
try (Scope scope = span.makeCurrent()) {
  SpanContext ctx = span.context();
  // process the event
  ...
} finally {
  span.end();
}
```

--------------------------------

### Extract OpenTracing context during consumption

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-opentracing/README.md

Use this snippet to extract the span context from event metadata and start a new span during event processing.

```java
SpanContext spCtx = OpenTracingHelper.mapToSpanContext(tracer, event.getMetadata().getSpanCtx());
Span span = tracer.buildSpan("sample-consuming-operation").asChildOf(spCtx).start();
try (Scope scope = tracer.activateSpan(span)) {
  SpanContext ctx = span.context();
  // process the event
  ...
} finally {
  span.finish();
}
```

--------------------------------

### Run full build and view aggregated coverage

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Execute the full test suite while skipping end-to-end tests, then open the aggregated coverage report.

```sh
./gradlew check -Pe2e.skip
open build/reports/jacoco/jacocoAggregateReport/html/index.html
```

--------------------------------

### Create Event Types

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-example/README.md

Executes the script to set up the necessary event types in the Nakadi instance.

```sh
fahrschein-example/scripts/create-event-types.sh
```

--------------------------------

### Publishing events with NakadiClient

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Initializes the NakadiClient with a request factory and access token provider, then publishes batches of events in a loop.

```java
// event types
String ORDER_CREATED = "order-created";

// Configure client. It is mandatory to choose the RequestFactory implementation (we recommend the JavaNetRequestFactory) and request encoding (we recommend compression via GZIP or ZSTD) 
// PlatformAccessTokenProvider provides a token by reading from the filesystem's CREDENTIALS_DIR 
NakadiClient nakadiClient = NakadiClient.builder(NAKADI_URI, new JavaNetRequestFactory(ContentEncoding.GZIP))
        .withAccessTokenProvider(new PlatformAccessTokenProvider())
        .build();

// Produce events in batches
while (true) {
    List<OrderCreatedEvent> events = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
        events.add(new OrderCreatedEvent(metadata, "123", Money.of(123, "EUR"), "1234", "paypal"));
    }
    LOG.info("publishing {} events", events.size());
    // Publish
    nakadiClient.publish(ORDER_CREATED, events);
}
```

--------------------------------

### Run local CVE scanning

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Execute dependency check to identify vulnerabilities and open the resulting report.

```sh
./gradlew :dependencyCheckAggregate
open build/reports/dependency-check-report.html
```

--------------------------------

### Configure Fahrschein via YAML

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-spring-boot-starter/README.md

Define global defaults and specific consumer configurations in your application.yml file to set up Nakadi clients.

```yaml
fahrschein:
  defaults:
    nakadi-url: http://localhost:8080
    application-name: example-application
    consumer-group: example-consumergroup
    stream-parameters:
      batch-limit: 1000
      max-uncommitted-events: 5000
      batch-flush-timeout: 5
    authorizations:
      anyReader: false
      admins:
        services:
          - some_service_identifier
        users:
          - user_6
          - user_2
        teams:
          - team_id_1
          - team_id_2
      readers:
        services:
          - some_other_service_identifier
    oauth:
      enabled: true
      credentials-directory: /meta/credentials
      access-token-id: nakadi
    record-metrics: true
    http:
      content-encoding: gzip
    backoff:
      enabled: true
  consumers:
    article-01:
      topics:
      - 'article-example_01924c48-AAAA-40c2-DDDD-ab582e6db6f4_ms'
      stream-parameters:
        batch-limit: 5
        max-uncommitted-events: 20
      backoff:
        enabled: false
    article-02:
      topics:
      - 'article-example_00f2a393-AAAA-4fc0-DDDD-86e454e6dfa3_ag'
    article-03:
      topics:
      - 'article-example_091dcbdd-AAAA-4f39-DDDD-324eb4599df0_io'
      - 'article-example_7ce94f55-AAAA-4416-DDDD-bf34193a47e8_co'
      - 'article-example_ebf57ebf-AAAA-4ebd-DDDD-6ad519073d2a_us'
      stream-parameters:
        batch-limit: 30
        max-uncommitted-events: 500
```

--------------------------------

### Configure Micrometer Metrics

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Integrate Micrometer metrics by providing a MeterRegistry to the stream configuration.

```java
MeterRegistry registry = new SimpleMeterRegistry();

nakadiClient.stream(subscription)
 .withMetricsCollector(new MicrometerMetricsCollector(registry))
```

--------------------------------

### Publish artifacts to local Maven repository

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Preview release artifacts locally by publishing to the local Maven repository without signing.

```sh
./gradlew publishToMavenLocal -xsignMavenPublication
```

--------------------------------

### Simplify event consumption with OpenTracingWrapper

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-opentracing/README.md

Use the wrapper class to automate span lifecycle management during event consumption.

```java
OpenTracingWrapper wrapper = new OpenTracingWrapper(tracer, "sample-consuming-operation");
wrapper.process(event, event -> doTheRealWork(event));
```

--------------------------------

### Add Maven Dependency

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-spring-boot-starter/README.md

Include this dependency in your project's pom.xml file to use the Fahrschein Spring Boot starter.

```xml
    <dependency>
        <groupId>org.zalando</groupId>
        <artifactId>fahrschein-spring-boot-starter</artifactId>
        <version>${version}</version>
    </dependency>
```

--------------------------------

### Simplify event consumption with OpenTelemetryWrapper

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-opentelemetry/README.md

Use the wrapper class to automate span lifecycle management during event processing.

```java
OpenTelemetryWrapper wrapper = new OpenTelemetryWrapper(tracer, "sample-consuming-operation");
wrapper.process(event, event -> doTheRealWork(event));
```

--------------------------------

### Add Spring Framework dependency

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Include the fahrschein-http-spring artifact to adapt other clients like OkHttp3.

```xml
<dependency>
    <groupId>org.zalando</groupId>
    <artifactId>fahrschein-http-spring</artifactId>
    <version>${fahrschein.version}</version>
</dependency>
```

--------------------------------

### Configure SpringRequestFactory with OkHttp3

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Wrap an OkHttp3 client using the SpringRequestFactory abstraction.

```java

ClientHttpRequestFactory clientHttpRequestFactory = new OkHttp3ClientHttpRequestFactory();
RequestFactory requestFactory = new SpringRequestFactory(clientHttpRequestFactory, ContentEncoding.GZIP);

NakadiClient nakadiClient = NakadiClient.builder(NAKADI_URI, requestFactory)
        .withAccessTokenProvider(new PlatformAccessTokenProvider())
        .build();
```

--------------------------------

### Subscribe to Event Types

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Create an event listener and configure the NakadiClient to stream events using the Subscription API.

```java
String eventName = "sales-order-placed";

// Create a Listener for our event
Listener<SalesOrderPlaced> listener = events -> {
    for (SalesOrderPlaced salesOrderPlaced : events) {
        LOG.info("Received sales order [{}]", salesOrderPlaced.getSalesOrder().getOrderNumber());
    }
};

// Configure client. It is mandatory to choose the RequestFactory implementation (we recommend the JavaNetRequestFactory) and request encoding (we recommend compression via GZIP or ZSTD) 
// PlatformAccessTokenProvider provides a token by reading from the filesystem's CREDENTIALS_DIR 
NakadiClient nakadiClient = NakadiClient.builder(NAKADI_URI, new JavaNetRequestFactory(ContentEncoding.GZIP))
        .withAccessTokenProvider(new PlatformAccessTokenProvider())
        .build();

// Create subscription using the Subscription API.
// The authorization section is mandatory. We recommend setting your team as admin
// and your application as reader.
Subscription subscription = nakadiClient.subscription(applicationName, eventName)
        .withAuthorization(authorization()
            .addAdmin("team", "your_team")
            .addReader("service", "stups_your_application").build())
        .subscribe();

// Define stream parameters, like batch size. For more details and defaults see Nakadi's API definition 
StreamParameters streamParameters = new StreamParameters()
                .withBatchLimit(25);

// Start streaming. The listen call will block and automatically reconnect on IOException
nakadiClient.stream(subscription)
        .withStreamParameters(streamParameters)
        .listen(SalesOrderPlaced.class, listener);
```

--------------------------------

### Add Apache HttpClient dependency

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Include the fahrschein-http-apache artifact to use the Apache HttpComponents implementation.

```xml
<dependency>
    <groupId>org.zalando</groupId>
    <artifactId>fahrschein-http-apache</artifactId>
    <version>${fahrschein.version}</version>
</dependency>
```

--------------------------------

### Add JDK11 HttpClient dependency

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Include the fahrschein-http-jdk11 artifact to use the Java 11+ HttpClient implementation.

```xml
<dependency>
    <groupId>org.zalando</groupId>
    <artifactId>fahrschein-http-jdk11</artifactId>
    <version>${fahrschein.version}</version>
</dependency>
```

--------------------------------

### Run unit tests and view coverage

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Execute tests for a specific subproject and open the generated JaCoCo coverage report.

```sh
./gradlew fahrschein-http-api:check
open fahrschein-http-api/build/reports/jacoco/test/html/index.html
```

--------------------------------

### Configure HttpComponentsRequestFactory

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Customize the Apache HttpClient for connection pooling and timeout settings.

```java
RequestConfig config = RequestConfig.custom().setSocketTimeout(readTimeout)
                                                   .setConnectTimeout(connectTimeout)
                                                   .setConnectionRequestTimeout(connectTimeout)
                                                   .build();

CloseableHttpClient httpClient = HttpClients.custom()
                                                  .setConnectionTimeToLive(readTimeout, TimeUnit.MILLISECONDS)
                                                  .disableAutomaticRetries()
                                                  .setDefaultRequestConfig(config)
                                                  .disableRedirectHandling()
                                                  .setMaxConnTotal(20)
                                                  .setMaxConnPerRoute(20)
                                                  .build();

RequestFactory requestFactory = new HttpComponentsRequestFactory(httpClient, ContentEncoding.GZIP);

NakadiClient nakadiClient = NakadiClient.builder(NAKADI_URI, requestFactory)
        .withAccessTokenProvider(new PlatformAccessTokenProvider())
        .build();
```

--------------------------------

### Override dependency versions via command line

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Use project properties to test specific dependency versions during the build process.

```sh
./gradlew check -Pjackson.version=2.9.0
```

--------------------------------

### Skip end-to-end tests

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Run the build process while explicitly excluding end-to-end tests.

```sh
./gradlew check -Pe2e.skip
```

--------------------------------

### Configuring automatic publishing retries

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Enables automatic retry mechanisms using a specific strategy and backoff configuration to handle partial batch failures.

```java

NakadiClient nakadiClient = NakadiClient.builder(NAKADI_URI, new JavaNetRequestFactory(ContentEncoding.GZIP))
        .withPublishingRetryAndBackoffStrategy(PublishingRetryStrategies.FAILED_ONLY, new ExponentialBackoffStrategy()) // recommended retry configuration
        .build();

nakadiClient.publish("foobar", List.of(
        new SomeEvent("eid1", new Metadata("eid1",OffsetDateTime.now())),
        new SomeEvent("eid2", new Metadata("eid2",OffsetDateTime.now())),
        new SomeEvent("eid3", new Metadata("eid3",OffsetDateTime.now())))
);
```

--------------------------------

### Implement Custom Authorization Provider

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Override the AuthorizationProvider interface to define a custom authorization flow. Pass the implementation to the NakadiClient builder using withAccessTokenProvider.

```java
//Create custom authorization
class CustomAuthorization implements AuthorizationProvider{

    @Override
    public String getAuthorizationHeader() throws IOException {
        return "your-secret-token";
    }
}

NakadiClient nakadiClient = NakadiClient.builder(NAKADI_URI, new SimpleRequestFactory(ContentEncoding.IDENTITY))
        .withAccessTokenProvider(new CustomAuthorization())
        .build();
```

--------------------------------

### Inject OpenTelemetry context into event metadata

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-opentelemetry/README.md

Use this snippet when creating a new Nakadi event to propagate the current OpenTelemetry context via metadata.

```java
Metadata md = new Metadata(UUID.randomUUID().toString(), OffsetDateTime.now(),
    "sample-flow-id", OpenTelemetryHelper.currentContextToMap());
```

--------------------------------

### Subscribe Using Existing Subscription ID

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Use a pre-existing subscription ID to stream events without creating a new subscription.

```java
// Create subscription using an existing subscription ID
Subscription subscription = nakadiClient
        .subscription("application-name",eventName)
        .subscribe("subscription-id");

// Start streaming, the listen call will block and automatically reconnect on IOException
nakadiClient.stream(subscription)
        .listen(SalesOrderPlaced.class, listener);
```

--------------------------------

### Inject OpenTracing context into event metadata

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/fahrschein-opentracing/README.md

Use this snippet when publishing an event to inject the active span context into the Nakadi event metadata.

```java
SpanContext ctx = tracer.activeSpan().context();
Metadata md = new Metadata(UUID.randomUUID().toString(), OffsetDateTime.now(),
    "sample-flow-id", OpenTracingHelper.spanContextToMap(tracer, ctx));
```

--------------------------------

### Coordinate Partition Consumption

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Use PartitionManager to lock partitions across multiple application nodes to ensure exclusive consumption.

```java
@Scheduled(fixedDelay = 60*1000L)
public void readSalesOrderPlacedEvents() throws IOException {
    String lockedBy = ... // host name or another unique identifier for this node
    List<Partition> partitions = nakadiClient.getPartitions(eventName);
    Optional<Lock> optionalLock = partitionManager.lockPartitions(eventName, partitions, lockedBy);

    if (optionalLock.isPresent()) {
        Lock lock = optionalLock.get();
        try {
            nakadiClient.stream(eventName)
                    .withLock(lock))
                    .listen(SalesOrderPlaced.class, listener);
        } finally {
            partitionManager.unlockPartitions(lock);
        }
    }
}
```

--------------------------------

### Add Fahrschein Maven Dependency

Source: https://github.com/zalando-nakadi/fahrschein/blob/main/README.md

Include this dependency in your project's pom.xml to use the Fahrschein library.

```xml
<dependency>
    <groupId>org.zalando</groupId>
    <artifactId>fahrschein</artifactId>
    <version>${fahrschein.version}</version>
</dependency>
```
Tokens:
Raw