# Configure message format for Kafka outbound

The `format` section of the `/etc/aerospike-kafka-outbound/aerospike-kafka-outbound.yml` specifies the serialization format of the data output. The supported formats are:

-   [Avro](#avro)
-   [Kafka Avro](#kafka-avro)
-   [Flat JSON](#flat-json)
-   [JSON](#json)
-   [MessagePack](#messagepack)
-   [Custom](#custom)

## Aerospike record metadata

The Aerospike record metadata has the following fields

| Metadata | Type | Description | Present in |
| --- | --- | --- | --- |
| `msg` | string | Write/Delete command. | Delete and Write |
| `namespace` | string | Namespace of the Aerospike record. | Delete and Write |
| `set` | string | Set of the Aerospike record. | Delete and Write |
| `userKey` | long, double, bytes or string | User key of the Aerospike record. | Write (only if user key is stored on server) |
| `digest` | bytes | Digest of the Aerospike record. | Delete and Write |
| `gen` | int | Generation of the Aerospike record. | Write |
| `lut` | int | Last update time of the Aerospike record. | Write |
| `exp` | int | Expiry of the Aerospike record. | Write |
| `durable` | boolean | Whether the delete is durable. | Delete |

::: caution
-   HyperLogLog bins are forwarded as is, as bytes or (for string-based formats) as Base64-encoded strings.
:::

## Avro

Specifies that the data be serialized as Avro format. The Avro schema can be a map or a record. In case of a map, just specifying the type of the map values is sufficient. In the case of a record, the exact field names and types must be specified.

| Option | Required | Default | Expected value | Description |
| --- | --- | --- | --- | --- |
| `mode` | yes |  | `avro` | Selects Avro format. |
| `schema` | no |  |  | Schema of the data. |
| `schema-file` | no |  |  | File containing the schema of the data. |
| `stringify-map-keys` | no | `true` |  | Whether the numeric keys in CDT maps should be converted to strings. See [Stringify Map Keys in Avro](https://aerospike.com/docs/connectors/streaming/kafka/outbound/formats/avro-serialization-format#stringify-map-keys-in-avro) |

One of `schema` or `schema-file` has to be specified.

## Kafka Avro

Specifies that the data be serialized with [KafkaAvroSerializer](https://github.com/confluentinc/schema-registry/blob/master/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroSerializer.java). The schema is expected to be maintained in a [schema registry](https://docs.confluent.io/current/schema-registry/index.html)

| Option | Required | Default | Expected value | Description |
| --- | --- | --- | --- | --- |
| `mode` | yes |  | `kafka-avro` | Selects Kafka Avro format. |
| `schema` | no |  |  | Schema of the data. |
| `schema-file` | no |  |  | The file containing the schema of the data. |
| `registry-topic` | no |  |  | Topic name used to generate the schema name in the schema registry. |
| `metadata-key` | no |  |  | Metadata will be inserted into this field of the record if specified, else metadata won’t be included in writes. |
| `props` | yes |  |  | Map of properties to initialize the KafkaAvroSerializer with. Valid values are as specified in [KafkaAvroSerializerConfig](https://github.com/confluentinc/schema-registry/blob/master/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroSerializer.java). |
| `stringify-map-keys` | no | `true` |  | Whether the numeric keys in CDT maps should be converted to strings. See [Stringify Map Keys in Avro](https://aerospike.com/docs/connectors/streaming/kafka/outbound/formats/avro-serialization-format#stringify-map-keys-in-avro) |

The writes and deletes are written with different schemas. Since only a single schema can be registered with `TopicNameStrategy` for a topic, this conflicts with accommodating both write and delete schemas. Hence `TopicNameStrategy` strategy is disallowed for `value.subject.name.strategy` value in the props.

## Flat JSON

Specifies that the data be serialized to [Flat JSON](https://aerospike.com/docs/connectors/streaming/kafka/outbound/formats/flat-json-serialization-format) format. The `type` information of bins is not part of the output.

| Option | Required | Default | Expected value | Description |
| --- | --- | --- | --- | --- |
| `mode` | yes |  | `flat-json` | Selects Flat JSON format. |
| `metadata-key` | yes |  |  | Aerospike record [metadata](#aerospike-record-metadata) will be inserted into this top level key. |
| `stringify-map-keys` | no | `true` |  | Whether the numeric keys in CDT maps should be converted to strings. See [Stringify Map Keys](#stringify-map-keys) |

#### Example

```yaml
format:

  mode: flat-json

  metadata-key: metadata
```

## JSON

Specifies that the data be serialized to [JSON](https://aerospike.com/docs/connectors/streaming/kafka/outbound/formats/json-serialization-format) format with the `type` information of bins.

| Option | Required | Default | Expected value | Description |
| --- | --- | --- | --- | --- |
| `mode` | yes |  | `json` | Selects JSON format. |
| `stringify-map-keys` | no | `true` |  | Whether the numeric keys in CDT maps should be converted to strings. See [stringify map keys](#stringify-map-keys) |

#### Stringify Map Keys

Aerospike supports `string`, numeric types, and `binary` for map keys. JSON supports only `string` keys. If a map key is a numeric type and if `stringify-map-keys` is set to `true`, that key is converted to a `string`.

In all other cases, a map with non-`string` keys cannot be parsed by the connector, and the connector throws an error.

#### Example

```yaml
format:

  mode: json
```

## MessagePack

Specifies that the data be serialized to [MessagePack](https://aerospike.com/docs/connectors/streaming/kafka/outbound/formats/messagepack-serialization-format) format with the `type` information of bins.

| Option | Required | Expected value | Description |
| --- | --- | --- | --- |
| `mode` | yes | `message-pack` | Selects message pack format. |

#### Example

```yaml
format:

  mode: message-pack
```

## Custom

Specifies that the data be serialized with user-provided custom code. See [Format Transform](https://aerospike.com/docs/connectors/streaming/common/outbound-message-transformer#format-transformer).

#### Example

```yaml
format:

  mode: custom

  class: com.aerospike.connect.outbound.example.Formatter
```