# Specify Pulsar topics

The `topics` section of the `aerospike-pulsar-inbound.yml` file tells the connector which Pulsar topics to consume messages from and how the data in messages corresponds to the data in your Aerospike database.

## Example

Here is an example of a `topics` section that specifies two topics to consume from. The names of the topics are `users` and `products`.

```yaml
topics:

  users:

    invalid-record: ignore

    mapping:

      namespace:

        mode: static

        value: users

      set:

        mode: dynamic

        source: value-field

        field-name: city

      key-field:

        source: key

      ttl:

        mode: dynamic

        source: value-field

        field-name: ttl

      bins:

        type: multi-bins

        map:

          name:

            source: value-field

            field-name: firstName

  products:

    invalid-record: kill-task

    mapping:

      namespace:

        mode: dynamic

        source: value-field

        field-name: category

      set:

        mode: dynamic

        source: value-field

        field-name: brand

      digest:

        source: value-field

        field-name: digest

      ttl:

        mode: dynamic

        source: value-field

        field-name: ttl

      bins:

        type: multi-bins

        map:

          name:

            source: value-field

            field-name: prod_name

          price:

            source: value-field

            field-name: prod_price
```

## Configure topics

For each topic that you name in the `topics` section, you can set properties in these sections:

| Option | Required | Default | Description |
| --- | --- | --- | --- |
| `aerospike-operation` | no |  | Specifies how to apply writes and deletes to your Aerospike database. See [Configuring Aerospike operations](#configure-aerospike-operations) for details. |
| `invalid-record` | yes | ignore | Action to take when a Pulsar record cannot be parsed or cannot be converted to a valid Aerospike record. Valid values are `ignore`, `kill-task`. This parameter is available from Pulsar inbound connector version **1.2.0**. |
| `mapping` | in some cases |  | Mapping of the Pulsar record to an Aerospike record. Required if you don’t implement `message-transformer` for the topic, or implement a `message-transformer` with [InboundMessage<K, M>](https://github.com/aerospike/aerospike-connect-inbound-sdk/blob/1.3.2/src/main/java/com/aerospike/connect/inbound/model/InboundMessage.java). See \[Map Pulsar messages to Aerospike records |
| \](/connectors/streaming/pulsar/inbound/configure/topics/#map-pulsar-messages-to-aerospike-records). |  |  |  |
| `message-transformer` | no |  | Configuration parameters for a message transformer. See [Aerospike Pulsar Inbound Message Transformer](https://aerospike.com/docs/connectors/streaming/pulsar/inbound/configure/message-transformer) for details. |

## Configure Aerospike operations

Use the `aerospike-operation` option to specify how to apply writes and deletes to your Aerospike database.

| Option | Required | Default | Description |
| --- | --- | --- | --- |
| `type` | no | `write` | Specifies the type of command to perform in Aerospike. Valid values are: `write`, `delete`. |
| `max-retries` | no | 2 | Specifies the maximum number of times to retry a transaction. |
| `total-timeout` | no | null | Specifies the number of milliseconds to wait before a transaction times out. If null, the `total-timeout` configured in the `aerospike` section is used. |
| `record-exists-action` | no | `update` | Specifies the action to take if a record already exists and `type` is set to `write`. Valid values are `update`, `update-only`, `replace`, `replace-only`, `create-only`. |
| `send-key` | no | `false` | Specifies to send a user-defined key, in addition to a hash digest. Applies only when the value for the property `type` is `write`. |
| `durable-delete` | no | `false` | Specifies to leave a tombstone for the record. Applies only when the value for the property `type` is `delete`. |
| `ignore-error-codes` | no | empty set | Consider a record as successfully processed if it throws [AerospikeException](https://github.com/aerospike/aerospike-client-java/blob/f46baecd7ace882244897c50d88b688fd545f5e2/client/src/com/aerospike/client/AerospikeException.java) with the given [ResultCode](https://github.com/aerospike/aerospike-client-java/blob/8f3d217237b280c1bc0420c58efa63428f8625bd/client/src/com/aerospike/client/ResultCode.java). |

### Examples

```yaml
# An aerospike write operation with all relevant options.

aerospike-operation:

  type: write

  max-retries: 3

  total-timeout: 0

  record-exists-action: update-only

  send-key: true

  ignore-error-codes:

    - 22

    - 13
```

```yaml
# An aerospike delete operation with all relevant options.

aerospike-operation:

  type: delete

  max-retries: 3

  total-timeout: 0

  durable-delete: true

  ignore-error-codes:

    - 22

    - 13
```

## Map Pulsar messages to Aerospike records

Use the `mapping` option to specify how to map Pulsar messages in a topic to Aerospike records.

| Option | Required | Description |
| --- | --- | --- |
| `namespace` | yes | Specifies whether the name of the namespace is set in the configuration file or in messages. See [Specify `namespace`, `set`, and `ttl`](#specify-namespace-set-and-ttl) for details. |
| `set` | no | Specifies whether the name of the set is set in the configuration file or in messages. See [Specify `namespace`, `set`, and `ttl`](#specify-namespace-set-and-ttl) for details. |
| `ttl` | no | Specifies whether the time to live is set in the configuration file or in messages. See [Specify `namespace`, `set`, and `ttl`](#specify-namespace-set-and-ttl) for details. |
| `bins` | yes | Mapping of incoming message to Aerospike bins. See [Mapping fields in messages to bins in Aerospike records](#mapping-fields-in-messages-to-bins-in-aerospike-records) for details. |
| `key-field` | no\* | The Field selection configuration value for the Aerospike record key. See [Key-field and Digest config](#key-field-and-digest-config) for details. |
| `digest` | no\* | The Field selection configuration value for the Aerospike record digest. The field has to be a valid [Aerospike RIPEMD-160](https://support.aerospike.com/s/article/FAQ-How-primary-keys-and-digests-are-used-in-Aerospike) digest. The digest field should be plain bytes or a Base64 encoded string. See [Key-field and Digest config](#key-field-and-digest-config) for details. |

Either `key-field` or `digest` has to be specified in a `mapping`.

### Key-field and Digest config

Use these two options to generate `key-field` or `digest`.

| Option | Required | Description |
| --- | --- | --- |
| `source` | yes | The source of the value. |
| `field-name` | no | Name of the field to extract the value from. Only applies when source is `value-field`. |

#### Possible values for the `source` option

| Value | Description |
| --- | --- |
| `key` | Use the whole of the source record key as the value. |
| `value` | Use the whole of the source record value as the value. |
| `value-field` | Extract a value from the source record value. |

## Specify `namespace`, `set`, and `ttl`

Configuration to obtain values for `namespace`, `set`, and `ttl` requires definitions for the following properties:

| Option | Required | Description |
| --- | --- | --- |
| `mode` | yes | How the value of a given field should be obtained. The possible options are `static` and `dynamic`. |

### Static: Specify the values in the configuration file

To set a static or unchanging name for the namespace, name for the set, or value for the time to live for an Aerospike record, you can use this option:

| Option | Required | Description |
| --- | --- | --- |
| value | yes | The static name or value. |

This example sets the name of a namespace to `east` for all records that an inbound connector creates from the messages that it receives:

```yaml
mapping:

  ...

  namespace:

    mode: static

    value: east

  ...
```

This example sets the time to live for all records that an inbound connector creates from the messages that it receives:

```yaml
mapping:

  ...

  ttl:

    mode: static

    value: -1

  ...
```

Choose from among the following values when setting the time to live for Aerospike records:

| Value | Description |
| --- | --- |
| `-2` | Specifies not to reset the time to live when a record is updated. |
| `-1` | Specifies that the time to live never expires. |
| `0` | Specifies to set the time to live to the default that is configured for the namespace in the Aerospike database. |
| Integer greater than 0 | Specifies by default the time to live in seconds. However, you can use one of the following suffixes to specify a different unit of time: `M` for minutes, `H` for hours, `D` for days. |

### Dynamic: Specify message fields from which to obtain values

Select a value dynamically from the Pulsar record.

| Option | Required | Description |
| --- | --- | --- |
| `source` | yes | The source of the value. |
| `field-name` | no | Name of the field to extract the value from. Only applies when source is `value-field`. |

#### Possible values for the `source` option

| Value | Description |
| --- | --- |
| `key` | Use the whole of the source record key as the value. |
| `value` | Use the whole of the source record value as the value. |
| `value-field` | Extract a value from the source record value. |

#### Example

```yaml
mapping:

  ...

  namespace:

    mode: dynamic

    source: value-field

    field-name: category

  set:

    mode: dynamic

    source: value-field

    field-name: brand

  ttl:

    mode: dynamic

    source: value-field

    field-name: validForSec

  ...
```

## Mapping fields in messages to bins in Aerospike records

For topics that you list in the `topics` section, you must map fields from messages to bins in Aerospike records.

| Option | Required | Description |
| --- | --- | --- |
| type | yes | Number of bins that exist in Aerospike records that correspond to messages in the topic. Valid values are `single-bin`, `multi-bins`. |

### Single bin

Use these two options to specify the source of the value to write to single-bin Aerospike records.

| Option | Required | Description |
| --- | --- | --- |
| `source` | yes | The source of the value. |
| `field-name` | no | Name of the field to extract the value from. Only applies when source is `value-field`. |

#### Possible values for the `source` option

| Value | Description |
| --- | --- |
| `key` | Use the whole of the source record key as the value. |
| `value` | Use the whole of the source record value as the value. |
| `value-field` | Extract a value from the source record value. |

#### Examples

```yaml
# A single bin Aerospike record with the bin value extracted from the Pulsar

# record field "name".

bins:

  type: single-bin

  source: value-field

  field-name: name
```

```yaml
# A single bin Aerospike record with the bin value equal to the whole of the

# Pulsar record value.

bins:

  type: single-bin

  source: value
```

### Multiple bins

Use these options to specify the source of the values to write to Aerospike records that are comprised of multiple bins.

| Option | Required | Default | Description |
| --- | --- | --- | --- |
| `all-value-fields` | no | false | Indicates if all fields from the Pulsar record value should be converted to Aerospike record bins. |
| `map` | no |  | Map of the Aerospike record bin name to the source of the bin value. Use this option only when `all-value-fields` is false. |
| `source` | yes |  | Source of the value. Used with the `map` option. |
| `field-name` | no |  | Name of the field to extract the value from. Only applies when source is `value-field`. Used with the `map` option. |

#### Possible values for the `source` option

| Value | Description |
| --- | --- |
| `key` | Use the whole of the source record key as the value. |
| `value` | Use the whole of the source record value as the value. |
| `value-field` | Extract a value from the source record value. |

#### Examples

```yaml
# Write all the Pulsar record value fields as Aerospike record bins.

bins:

  type: multi-bins

  all-value-fields: true
```

```yaml
# Extract the values of Aerospike bins "name" and "city" from the Pulsar fields `firstName` and `residence`.

bins:

  type: multi-bins

  map:

    name:

      source: value-field

      field-name: firstName

    city:

      source: value-field

      field-name: residence
```

```yaml
# Use the whole of the Pulsar record value as the Aerospike bin "metadata".

bins:

  type: multi-bins

  map:

    metadata:

      source: value
```