Source: https://dev-docs.nebula-graph.io/v5.3.2/utilities/exchange/config/sources/kafka/
Documentation version: v5.3.2

# Kafka configurations

This topic introduces the configuration options for importing data from a Kafka data source. For information about the complete configuration file and options, see [Data import configurations](https://dev-docs.nebula-graph.io/v5.3.2/utilities/exchange/config/).

## Data source configurations

The following table lists the configuration options specific to the Kafka data source.

 | Option | Description | Type | Required | Default value
 | `type` | The type of the data source. In this case, set the value to `"kafka"`. | String | Yes | N/A
 | `server` | The address and port of the Kafka server. | String | Yes | N/A
 | `topic` | The Kafka topic from which data is retrieved. | String | Yes | N/A
 | `startingOffsets` | The starting offset for reading data from the Kafka topic. 
Valid values are `"earliest"` and `"latest"`. To read data from the earliest offset, set the value to `"earliest"`. To read data from the latest offset, set the value to `"latest"`. | String | No | `"earliest"`
 | `maxOffsetsPerTrigger` | The maximum number of offsets to process per trigger interval. | Int | No | `null`
 | `securityProtocol` | The security protocol used to communicate with the Kafka server. Valid values are `"PLAINTEXT"`, `"SASL_PLAINTEXT"`, `"SASL_SSL"`, and `"SSL"`. 
This option is required only when the Kafka server is authenticated. | String | No | `null`
 | `mechanism` | The security mechanism used for authentication. Valid values are `"GSSAPI"`, `"PLAIN"`, `"SCRAM-SHA-256"`, and `"SCRAM-SHA-512"`. 
This option is required only when the Kafka server is authenticated. | String | No | `null`
 | `kerberos` | Whether Kerberos authentication is enabled. 
This option must be set to `true` when the Kafka server is authenticated with Kerberos. | Bool | No | `false`
 | `kerberosServiceName` | The Kerberos service name. 
This option is required only when Kerberos authentication is enabled. | String | No | `null`

## Common source configurations

The following configuration options can also be specified as global settings in the section `sourcesCommon` of the configuration file:

- `type`
- `server`
- `startingOffsets`
- `maxOffsetsPerTrigger`
- `securityProtocol`
- `mechanism`
- `kerberos`
- `kerberosServiceName`

Note that the configurations on specific data sources override the global settings.

## Data source configuration example

Kafka source configurations

````
sources: [
  {
      type:"kafka"
      server: "127.0.0.1:9092"
      topic:"movie"
      startingOffsets: "earliest"
      maxOffsetsPerTrigger: 1000000
      # if your kafka server is authenticated, then you need to config the following four configs.
      securityProtocol: SASL_PLAINTEXT
      mechanism: GSSAPI
      kerberos: true
      kerberosServiceName: kafka
//      nodetypes: []
//      edgetypes: []
  }
]    

````

## Complete configuration example

Replace `<current-password>` with the current password for the specified account. If the password has changed, use the new password.

````
{
  taskType: import
  nebula: {
    graphAddr: "127.0.0.1:9669"
    graphName: "movie"
    user: "root"
    passwd: "<current-password>"
  }

  error: {
    path: "/tmp/errors"
    maxRecords: 10000
  }

  sourcesCommon:{
    type:"kafka"
    server: "127.0.0.1:9092"
    startingOffsets: "earliest"
    maxOffsetsPerTrigger: 1000000
  }

  writeCommon:{
    batchSize: 2000
    enableRealTimeQPSLog: true
    mode: "INSERT IGNORE"
  }

  sources: [
    # node type Actor
    {
      topic: actor
      pre_processes: {
          filter: ["birth_date!='\\N'"]
          concatConstant: [
            {
              field: "name"
              prefixValue: "test_"
              suffixValue: ""
            }
          ]
      }

      nodetypes: [
        {
          name: "Actor"
          sourceFields: [people_id,name,birth_date]
          nebulaFields: [id,name,birthDate]
        }
      ]
    }

    # node type Movie
    {
      topic: movie
      nodetypes: [
        {
          name: "Movie"
          sourceFields: [movie_id,name]
          nebulaFields: [id,name]
        }
      ]
    }

   # edge type Act
    {
      topic: act
      edgetypes: [
        {
          name: "Act"
          sourceKeys: [people_id]
          nebulaSrcKeys: [id]
          targetKeys: [movie_id]
          nebulaDstKeys: [id]
          sourceFields: []
          nebulaFields: []
        }
      ]
    }

  ]
}

````