# Kafka as a data source

> Learn how to make a Kafka Topic available for Orbital to query

Source: https://orbitalhq.com/docs/describing-data-sources/kafka

Orbital can both read data from a Kafka topic to stream data as part of a query, and write data
back to Kafka.

## Defining a connection to your Kafka broker
Kafka connections are stored in your `connections.conf` config file, under the `kafka` element.

The connection specifies how to connect to a broker - but not the specifics of actual topics.
(Those are specified on the `@KafkaOperation()` metadata on a taxi operation definition.)

```hocon
kafka {
   my-kafka-connection {
      connectionName=my-kafka-connection
      connectionParameters {
         brokerAddress="localhost:29092,localhost:39092"
         groupId="orbital"
      }
   }
   another-kafka-connection {
      connectionName=another-kafka-connection
      connectionParameters {
         brokerAddress="localhost:29092,localhost:39092"
         groupId="orbital"
      }
   }
}
```

The following configuration options are supported under the `connectionParameters`

| Config option | Purpose                                                                        |
|---------------|--------------------------------------------------------------------------------|
| `brokers`     | A comma-separated list of broker addresses when connecting to the Kafka broker |
| `groupId`     | The groupId to use when connecting to a topic                                  |

### Additional Kafka connection properties
In addition to `brokers` and `groupId` (which are shorthand for `bootstrap.servers` and `group.id` respectively), you can provide
any of the Kafka [consumer](https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html) and [producer](https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html) config settings.

For example, to define a secure connection:

```hocon
kafka {
   my-kafka-connection {
      connectionName=my-kafka-connection
      connectionParameters {
         brokerAddress="localhost:29092,localhost:39092"
         groupId="orbital"
         "security.protocol"="SASL_PLAINTEXT"
         "sasl.mechanism"="PLAIN"
         "sasl.jaas.config"="org.apache.kafka.common.security.plain.PlainLoginModule required username='myKafkaUser' password="${KAFKA_PASSWORD}";"
      }
   }
}
```


When using Kafka connections to consume data, specifying an appropriate `groupId` is important, and impacts the behaviour of how messages are received.  

If you are experiencing [long delays at the start of a query](#long-connection-times-when-starting-a-query), or [no results returned](#query-doesn-t-return-any-data-even-though-data-exists), you may need to modify your `groupId`.

You can set `groupId` in `connectionParameters` to specify the group id of your Orbital consumer.

```hocon
kafka {
   my-kafka-connection {
      connectionName=my-kafka-connection
      connectionParameters {
         brokerAddress="localhost:29092,localhost:39092"
         groupId="myGroupId"
      }
   }
}
```

Read more about consumer group id's in the [Kafka documentation](https://www.confluent.io/blog/configuring-apache-kafka-consumer-group-ids/) 

### Troubleshooting connection problems

#### Using env variables in auth settings
It's common in Kafka configs that usernames and passwords are embedded within settings, like `sasl.jaas.config`.

It's recommended that usernames and passwords are read from env variables, which can lead to tricky [HOCON concatenation issues](https://github.com/lightbend/config/issues/113).

The solution is to rely on HOCON concatenation. eg:

```HOCON
"sasl.jaas.config"="org.apache.kafka.common.security.plain.PlainLoginModule required username="${KAFKA_USERNAME}" password="${KAFKA_PASSWORD}";"
```

Pay close attention to how the `"` in the setting are paired - ie., no escaping, and no using `'` to nest inner values.

For example:

```hocon
kafka {
   my-kafka-connection {
      // ... omitted for brevity
      connectionParameters {
         "sasl.jaas.config"="org.apache.kafka.common.security.plain.PlainLoginModule required username='myKafkaUser' password="${KAFKA_PASSWORD}";"
      }
   }
}
```

#### Query doesn't return any data (even though data exists)
You might have more subscribers with the 'groupId' than the configured partition count for the topic.  
In this case, your Orbital instance won't get any data from the topic. Please set the groupId in your kafka and specify a new groupId:

```hocon
kafka {
   my-kafka-connection {
      connectionName=my-kafka-connection
      connectionParameters {
         brokerAddress="localhost:29092,localhost:39092"
         groupId="myGroupId"
      }
   }
}
```

#### Long connection times when starting a query
When multiple Orbital instances connect (and disconnect) to the same Kafka broker with the same `groupId` (such as the default setting of `groupId = Orbital`), this can trigger on the Kafka broker.

As a result, long delays are sometimes observed before data is streamed from Kafka, while the cluster is rebalancing.

Resolve this by setting the `groupId` for your kafka connection to a unique value in your connections.conf

## Querying a Kafka topic
A Kafka topic can be queried using a standard Taxi query, including enriching data
from Kafka with other services, and [joining multiple streams together](#joining-multiple-streams)

### Exposing a topic
Kafka topics are declared in taxi as simple Operations which 
return a Stream of data.

```taxi
// 1: Add the required imports
import com.orbitalhq.kafka.KafkaService
import com.orbitalhq.kafka.KafkaOperation

// 2: Annotate the service as a `KafkaService`.
@KafkaService( connectionName = "market-prices" )
service MyKafkaService {

   // 3: Annotate the operation as a `@KafkaOperation`
   @KafkaOperation( topic = "stockPrices", offset = "earliest" )
   // The topic is declared as a stream, and returns a Stream<>
   stream streamStockPrices:Stream<StockPrice>
}
```

 1. Add the required imports
 1. Annotate the service with `@KafkaService`
    * The `connectionName` parameter should match a connection defined in the [connections config file](#defining-a-connection-to-your-kafka-broker).
 1. Annotate the operation as a `@KafkaOperation`, specifying the topic and offset.
 1. The return type should be a `Stream<>` of the defined message type.

It's possible to use environment variables in these annotations, as described [here](/docs/deploying/configuring-orbital#environment-variables-in-annotations).

### Keys, Headers and Metadata
The message key, message headers, and kafka-specific metadata (such as offset, partition and timestamp) can be provided
into a Taxi model using annotations:

#### Message Key
The message key can be accessed using the `com.orbtalhq.kafka.KafkaMessageKey` annotation on a field:

```taxi
> import com.orbitalhq.kafka.KafkaMessageKey
  
  model Movie {
    // The key from Kafka will be read into the id property
>   @KafkaMessageKey
    id : MovieId inherits String
    title : Title inherits String
  }

  // Rest of the kafka topic declaration continues...
  @KafkaService( connectionName = "moviesConnection" )
  service MovieService {
    @KafkaOperation( topic = "movies", offset = "earliest" )
    stream streamMovieQuery:Stream<Movie>
  }
```

#### Kafka Metadata
Kafka metadata (such as offset, partition and timestamp) can be accessed using the  `com.orbitalhq.kafka.KafkaMessageMetadata`
annotation on a field.

`KafkaMessageMetadata` takes a single parameter, which is the metadata type you wish to read. Defined by the enum 
type `KafkaMetadataType`, the following values are defined:

```taxi
enum KafkaMetadataType {
  Partition,
  Offset,
  Timestamp,
  TimestampType
  Topic
}
```

For example:

```taxi
>   import com.orbitalhq.kafka.KafkaMessageMetadata
>   import com.orbitalhq.kafka.KafkaMetadataType
    
    model Movie {
>     @KafkaMessageMetadata(KafkaMetadataType.Offset)
      offset : Int
    
>     @KafkaMessageMetadata(KafkaMetadataType.Timestamp)
      timestamp : Long
    
>     @KafkaMessageMetadata(KafkaMetadataType.TimestampType)
      timestampType : String
    
>     @KafkaMessageMetadata(KafkaMetadataType.Partition)
      partition : Int

>     @KafkaMessageMetadata(KafkaMetadataType.Topic)
      topic : String

      // Other fields continue...
      title : Title inherits String
    }
    
    // Rest of the kafka topic declaration continues...
    @KafkaService( connectionName = "moviesConnection" )
    service MovieService {
      @KafkaOperation( topic = "movies", offset = "earliest" )
      stream streamMovieQuery:Stream<Movie>
    }
```

#### Headers
Kafka supports including arbitrary message headers along with the message - which are often used
for things like correlation keys, etc.

These headers can be accessed using the `com.orbitalhq.kafka.KafkaHeader` annotation:

```taxi
   import com.orbitalhq.kafka.KafkaHeader
   
   model Movie {
>     @KafkaHeader("correlationId")
      correlationId : CorrelationId inherits String
      title : Title inherits String
   }

    // Rest of the kafka topic declaration continues...
   @KafkaService( connectionName = "moviesConnection" )
   service MovieService {
      @KafkaOperation( topic = "movies", offset = "earliest" )
      stream streamMovieQuery:Stream<Movie>
   }
```

### Controlling deserialization
Message deserialization is defined by the model type being exposed.
By default, models are expected to be JSON.

However, this can be controlled by annotating the model with a format annotation.

Two common formats - [Protobuf](/docs/describing-data-sources/protobuf/) and [Avro](/docs/data-formats/avro) are supported.

### Example queries

#### Streaming data from Kafka
```taxi
// Invokes the `streamStockPrices` stream declared above
stream { StockPrice }
```

#### Enrich data from Kafka with other data sources
Data from a Kafka topic can be projected to enrich it with data from
other sources.

Data requested that is not present on the Kafka payload is looked up
from other sources, using Orbital's standard projections.

```taxi
stream { StockPrice } as {
  ticker : StockTicker  // avaialble on the Kafka topic
  lastTradedPrice : LastTradedPrice // Looked up from another data source
}[]
```

#### Filtering kafka streams
This examples reads all messages from the Kafka topic, but only
emits those with a stock ticker of `AAPL` on the resulting stream:

```taxi
stream { StockPrice.filterEach( ( StockTicker ) -> StockTicker == 'AAPL' ) }
```

#### Streaming from Kafka to a database
Streams from Kafka can be inserted into a database (or any other writable source— such as
[Hazelcast](/docs/describing-data-sources/hazelcast#writing-data-to-hazelcast) or [Dynamo]()) using a mutating query.

As with all mutating queries, it's not
necessary for the data from Kafka to align with the format of the
data being written to the data destination.

Orbital will automatically adapt the query result to the
required persistence format, which may involve projections and even
calling additional services if needed.

```taxi
// First, ensure that your data destination exposes a writeable data source
// Full config omitted for brevity
service MyDatabaseService {
   @UpsertOperation
   write operation updateStockPrices(StockPriceSnapshot):StockPriceSnapshot
}

// Then, define a streaming query.
// In this example, the data format for StockPrice coming off of Kafka
// is different from the data being written to our database (StockPriceSnapshot)
// so Orbital transforms the data automatically
stream { StockPrice }
call MyDatabaseService::updateStockPrices
```

### Joining multiple streams
It is possible to join multiple Kafka streams together as described in [joining streams](/docs/querying/streaming-data#joining-streams)

### Overriding consumer group and offset
By default, the consumer group is defined on the operation:

```taxi
@KafkaService( connectionName = "moviesConnection" )
service MovieService {
  @KafkaOperation( topic = "movies", offset = "earliest" )
  stream streamMovieQuery:Stream<Movie>
}
```

and the consumer group is defined in your connections.conf:

```hocon
   kafka {
      my-kafka-connection {
>        connectionName="moviesConnection"
         connectionParameters {
            brokerAddress="localhost:29092,localhost:39092"
>           groupId="orbital"
         }
      }
   }
```

However, these can be overridden on a per-query basis, by using a `@StreamConsumer` query:

```taxi
>  // 1: don't forget the import
>  import com.orbitalhq.streams.StreamConsumer 
   
>  // changes the consumer group id on a Kafka subscription and offset
>  @StreamConsumer( id = "my-consumer" , offset = "latest" ) 
   query MySavedQuery {
      stream { Movie }
   }
```

Likewise, for an adhoc query:

```taxi
>  // 1: don't forget the import
>  import com.orbitalhq.streams.StreamConsumer 
   
>  // changes the consumer group id on a Kafka subscription and offset
>  @StreamConsumer( id = "my-consumer" , offset = "latest" ) 
   stream { Movie }
```

## Writing to a Kafka topic
To make a topic writable, declare a `write operation` in a Kafka service:

```taxi
// 1: Add the required imports
import com.orbitalhq.kafka.KafkaService
import com.orbitalhq.kafka.KafkaOperation

// 2: Annotate the service as a `KafkaService`.
@KafkaService( connectionName = "market-prices" )
service MyKafkaService {

   // ...other kafka topics omitted...

   // 3: Annotate the operation as a `@KafkaOperation`
   @KafkaOperation( topic = "stockPrices", offset = "earliest" )
   // The operation is declared as a write operation
   write operation publishStockPrice(StockPrice):StockPrice
}
```

### Examples

#### Writing a static value onto a Kafka topic
```taxi
given { stockPrice : StockPrice = 
  {
    symbol : 'AAPL',
    price : 12.00203,  
  } 
}
call MyKafkaService::publishStockPrice
```

#### Consuming from one Kafka topic, and writing to another topic
To stream data from a Kafka topic, enrich and republish

```taxi
@KafkaService( connectionName = "market-prices" )
service MyKafkaService {

   @KafkaOperation( topic = "stockPrices" )
   stream prices : Stream<StockPrice>

   @KafkaOperation( topic = "enrichedPrices" )
   write operation publishEnrichedPrices(EnrichedStockPrice):EnrichedStockPrice 
} 
```

The following query will consume from the `stockPrices` topic, and for each message,
transform to an `EnrichedStockPrice`, invoking any other services required to inject required data.

```taxi
stream { StockPrice }
// The input parameter to publishEnrichedPrices
// is a EnrichedStockPrice, so each incoming 
// StockPrice message is transformed to a
// EnrichedStockPrice payload, and published onto the
// enrichedPrices topic
call MyKafkaService::publishEnrichedPrices
```

#### Building a REST API that publishes to Kafka
This is a full example, where we create an HTTP endpoint accepting a `POST` request
with a ticker symbol.

```taxi
type StockSymbol inherits String
// The inbound request sent over HTTP requesting a stock price
model StockPricePublicationRequest {
  ticker : StockSymbol
}

// The message we'll be publishing to Kafka
parameter model StockPriceUpdate {
  ticker : StockSymbol
  currentPrice : StockPrice 
}

closed model CurrentStockPrice {
  price : StockPrice
}

service PriceService {
  @HttpOperation(url="http://fakeurl/prices/{symbol}", method = "GET")
  operation getCurrentPrice(@PathVariable("symbol") symbol:StockSymbol):CurrentStockPrice  
}

@KafkaService( connectionName = "market-prices" )
service MyKafkaService {
   @KafkaOperation( topic = "stockPrices", offset = "earliest" )
   write operation publishStockPrice(StockPriceUpdate):StockPriceUpdate
}

@HttpOperation(path = "/api/q/publishStockPrice", method = "POST")
query MySavedQuery(@RequestBody request:StockPriceRequest) {
  given { request }
  call MyKafkaService::publishStockPrice
}
```

The above example works as follows:

 * A `POST` request is sent to `/api/q/publishStockPrice` with a body of:

```json
{ "ticker" : "AAPL" }
```

 * The query asks for `publishStockPrice` to be called, which means a `StockPriceUpdate` must be constructed
 * To build a `StockPriceUpdate`, the `currentPrice : StockPrice` is required, which is available from the `price` field of `CurrentStockPrice` object, returned from `getCurrentPrice`
 * A request to `http://fakeurl/prices/AAPL` is issued to discover the current stock price, returning:

```json
{ "price" : 117.34 }
```

 * Finally, we have enough information to build a `StockPriceRequest`, so the message is published to Kafka:

```json
{ "ticker" : "AAPL", "currentPrice" : 117.34 }
```