<commit_before>[![Build Status](https://travis-ci.org/zalando/nakadi.svg)](https://travis-ci.org/zalando/nakadi)
[![ReviewNinja](https://app.review.ninja/44234368/badge)](https://app.review.ninja/zalando/nakadi)
[![codecov.io](https://codecov.io/github/zalando/nakadi/coverage.svg?branch=nakadi-jvm)](https://codecov.io/github/zalando/nakadi?branch=nakadi-jvm)

[![Swagger API](http://online.swagger.io/validator?url=https://raw.githubusercontent.com/zalando/nakadi/nakadi-jvm/api/nakadi-event-bus-api.yaml)](http://online.swagger.io/validator?url=https://raw.githubusercontent.com/zalando/nakadi/nakadi-jvm/api/nakadi-event-bus-api.yaml)

# Nakadi Event Bus

The goal of the `nakadi` project (ნაკადი means `stream` in Georgian language) is to build an event bus infrastructure to:

*  enable convenient development of event-driven applications
*  securely and efficiently publish and consume events as easy as possible
*  abstract event exchange by a standardized RESTful [API](/api/nakadi-event-bus-api.yaml)

Some additional technical requirements that we wanted to cover by this architecture:

* event ordering guarantees
* fast (near real-time) event processing
* scalable and highly available architecture
* [STUPS](https://stups.io/) compatible

Additional topics, that we plan to cover in the near future are:

* discoverability of the resource structures flowing into the event bus
* centralized discovery service, that will use these capabilities to collect resource schema information for easy lookup by developers

> NOTE: it is not really clear if the resource schema discoverability service should be part of `nakadi` event bus

## What does the project already implement?

* [x] REST abstraction over Kafka-like queues
* [ ] support of event filtering per subscriptions
* streaming/batching of events to/from the clients
  * [x] creation of topics
  * [x] low-level interface
    * manual client side partition management is needed
    * no support of commits
  * [ ] high-level interface
    * automatic redistribution of partitions between consuming clients
    * commits should be issued to move server-side cursors

# Running it locally

## To run the project locally

Simple Nakadi startup:

```sh
./gradlew startDockerContainer
```

It will start a docker container with all dependencies and another docker container running Nakadi itself. Please be
aware that the ports 8080 (Nakadi), 5432 (PostgreSQL), 9092 (Kafka) and 2181 (Zookeeper) are needed and must not be
blocked by another application.

To stop the running Nakadi again:

```sh
./gradlew stopAndRemoveDockerContainer
```

## Full development pipeline:

    build -> ut/it tests (depends on access to a Kafka backend) -> docker (builds docker image) -> api-tests (runs tests against the docker image)

# Usage

## Create new event type (business event)

```sh
curl --request POST \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     http://localhost:8080/event-types -d '{
  "name": "order.ORDER_RECEIVED",
  "owning_application": "order-service",
  "category": "business",
  "partitioning_key_fields": [],
  "schema": {
    "type": "json_schema",
    "schema": "{ \"properties\": { \"order_number\": { \"type\": \"string\" } } }"
  }
}'
```

## Get existing event types

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     http://localhost:8080/event-types
```

## Get event type schema

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     http://localhost:8080/event-types/order.ORDER_RECEIVED
```

## Get all partitions for event type

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     http://localhost:8080/event-types/order.ORDER_RECEIVED/partitions
```

## Get single partition for event type

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     http://localhost:8080/event-types/order.ORDER_RECEIVED/partitions/0
```

## Publish event

```sh
curl --request POST \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     http://localhost:8080/event-types/order.ORDER_RECEIVED/events \
     -d '{ "order_number": "ORDER_ONE", "metadata": { "eid": "39ac3332-eb00-11e5-91fe-1c6f65464fc6", "occurred_at": "2016-03-15T23:47:15+01:00" } }'
```

## Receive event stream

Lookup the partition cursors with `/event-types/order.ORDER_RECEIVED/partitions` in order to provide a valid `X-Nakadi-Cursors` header.

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     --header 'X-Nakadi-Cursors:[{"partition": "0", "offset":"0"}]' \
     http://localhost:8080/event-types/order.ORDER_RECEIVED/events
```

The stream contains events together with the cursors, so that clients can remember which events have already been consumed and navigate through the stream. Example:

```sh
$ curl --request GET \
       --header "Authorization: Bearer $TOKEN" \<commit_msg>Added linebreaks to example curl calls<commit_after>[![Build Status](https://travis-ci.org/zalando/nakadi.svg)](https://travis-ci.org/zalando/nakadi)
[![ReviewNinja](https://app.review.ninja/44234368/badge)](https://app.review.ninja/zalando/nakadi)
[![codecov.io](https://codecov.io/github/zalando/nakadi/coverage.svg?branch=nakadi-jvm)](https://codecov.io/github/zalando/nakadi?branch=nakadi-jvm)

[![Swagger API](http://online.swagger.io/validator?url=https://raw.githubusercontent.com/zalando/nakadi/nakadi-jvm/api/nakadi-event-bus-api.yaml)](http://online.swagger.io/validator?url=https://raw.githubusercontent.com/zalando/nakadi/nakadi-jvm/api/nakadi-event-bus-api.yaml)

## Nakadi Event Bus

The goal of the `nakadi` project (ნაკადი means `stream` in Georgian language) is to build an event bus infrastructure to:

*  enable convenient development of event-driven applications
*  securely and efficiently publish and consume events as easy as possible
*  abstract event exchange by a standardized RESTful [API](/api/nakadi-event-bus-api.yaml)

Some additional technical requirements that we wanted to cover by this architecture:

* event ordering guarantees
* fast (near real-time) event processing
* scalable and highly available architecture
* [STUPS](https://stups.io/) compatible

Additional topics, that we plan to cover in the near future are:

* discoverability of the resource structures flowing into the event bus
* centralized discovery service, that will use these capabilities to collect resource schema information for easy lookup by developers

> NOTE: it is not really clear if the resource schema discoverability service should be part of `nakadi` event bus

### What does the project already implement?

* [x] REST abstraction over Kafka-like queues
* [ ] support of event filtering per subscriptions
* streaming/batching of events to/from the clients
  * [x] creation of topics
  * [x] low-level interface
    * manual client side partition management is needed
    * no support of commits
  * [ ] high-level interface
    * automatic redistribution of partitions between consuming clients
    * commits should be issued to move server-side cursors

## Running it locally

### To run the project locally

Simple Nakadi startup:

```sh
./gradlew startDockerContainer
```

It will start a docker container with all dependencies and another docker container running Nakadi itself. Please be
aware that the ports 8080 (Nakadi), 5432 (PostgreSQL), 9092 (Kafka) and 2181 (Zookeeper) are needed and must not be
blocked by another application.

To stop the running Nakadi again:

```sh
./gradlew stopAndRemoveDockerContainer
```

### Full development pipeline:

    build -> ut/it tests (depends on access to a Kafka backend) -> docker (builds docker image) -> api-tests (runs tests against the docker image)

## Usage

### Create new event type (business event)

```sh
curl --request POST \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     http://localhost:8080/event-types -d '{
  "name": "order.ORDER_RECEIVED",
  "owning_application": "order-service",
  "category": "business",
  "partitioning_key_fields": [],
  "schema": {
    "type": "json_schema",
    "schema": "{ \"properties\": { \"order_number\": { \"type\": \"string\" } } }"
  }
}'
```

### Get existing event types

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     http://localhost:8080/event-types
```

### Get event type schema

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     http://localhost:8080/event-types/order.ORDER_RECEIVED
```

### Get all partitions for event type

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     http://localhost:8080/event-types/order.ORDER_RECEIVED/partitions
```

### Get single partition for event type

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     http://localhost:8080/event-types/order.ORDER_RECEIVED/partitions/0
```

### Publish event

```sh
curl --request POST \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     http://localhost:8080/event-types/order.ORDER_RECEIVED/events \
     -d '{ "order_number": "ORDER_ONE", "metadata": { "eid": "39ac3332-eb00-11e5-91fe-1c6f65464fc6", "occurred_at": "2016-03-15T23:47:15+01:00" } }'
```

### Receive event stream

Lookup the partition cursors with `/event-types/order.ORDER_RECEIVED/partitions` in order to provide a valid `X-Nakadi-Cursors` header.

```sh
curl --request GET \
     --header "Authorization: Bearer $TOKEN" \
     --header "Content-Type:application/json" \
     --header 'X-Nakadi-Cursors:[{"partition": "0", "offset":"0"}]' \
     http://localhost:8080/event-types/order.ORDER_RECEIVED/events
```

The stream contains events together with the cursors, so that clients can remember which events have already been consumed and navigate through the stream. Example:

```sh
$ curl --request GET \
       --header "Authorization: Bearer $TOKEN" \
{'fraction_non_alphanumeric': 0.12846273897776045, 'fraction_numerical': 0.01833788529067499, 'mean_word_length': 4.0557199211045365, 'pattern_counts': {'":': 32, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 18, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '999302', 'n_tokens_mistral': 3288, 'n_tokens_neox': 3080, 'n_words': 939}