KafkaV3

Details

ID	`KafkaV3`
Provider	Open Source
Category	Messaging and Conferencing
From Version	6.1.0
Docker Image	`demisto/vendors-sdk:1.0.0.10120494`
Supported Modules	Agentix XSIAM

README

Use the Kafka integration to manage messages and partitions and to fetch Kafka messages to create incidents in Cortex XSOAR.

This integration was integrated and tested with version 2.7.1 of Kafka.

This integration is fully compatible with the Kafka v2 integration.

Configure Kafka v3 in Cortex

Parameter	Description	Required
CSV list of Kafka brokers to connect to, e.g., 172.16.20.207:9092,172.16.20.234:9093		True
Consumer group ID	This group ID will be used when fetching incidents and preforming consumer commands. If not set the group id ‘xsoar_group’ will be used.	False
Use TLS for connection		False
Use SASL PLAIN for connection (using SSL)
Trust any certificate (not secure)		False
CA certificate of Kafka server (.cer)		False
Client certificate (.cer)		False
Client certificate key (.key)		False
Client certificate key password (if required)		False
SASL PLAIN Username		False
SASL PLAIN Password		False
Topic to fetch incidents from (Required for fetch incidents)		False
CSV list of partitions to fetch messages from		False
Offset to fetch messages from (Exclusive)	The initial offset to start fetching from, not including the value set (e.g., if 3 is set, the first event that will be fetched will be from offset 4). If you want to start from the earliest or latest, type in ‘earliest’ or ‘latest’.	False
Maximum number of messages to fetch		False
Stop consuming upon timeout	When fetching a significant number of messages (100+), it’s advisable to halt message consumption upon timeout. This ensures that the fetch terminates if no messages are received after a specified duration, instead of requesting messages until reaching the maximum number of messages to fetch.	False
Fetch incidents		False
Incident type		False
Maximum number of bytes per message	The maximum number of message bytes to retrieve in each attempted fetch request. Should be in multiples of 1024. If the fetching process takes a long time, consider increasing this value. Default is ‘1048576’.	False
Schema Registry URL		False
Schema Registry Username		False
Schema Registry Password		False

Commands

You can execute these commands from the CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.

kafka-print-topics

Prints all topics and their partitions.

Base Command

kafka-print-topics

Input

Argument Name	Description	Required
include_offsets	Whether to include the first and last offset for a topic, when printing a list of topics and partitions. Possible values are: true, false. Default is true.	Optional

Context Output

Path	Type	Description
Kafka.Topic.Name	String	Kafka topic name.
Kafka.Topic.Partitions.ID	Number	Topic partition ID.
Kafka.Topic.Partitions.EarliestOffset	Number	Topic partition earliest offset.
Kafka.Topic.Partitions.LatestOffset	Number	Topic partition latest offset.

Command Example


#### Context Example

{
“Kafka”: {
“Topic”: [
{
“Name”: “test-topic1”,
“Partitions”: [
{
“ID”: 0
}
]
},
{
“Name”: “test-topic2”,
“Partitions”: [
{
“ID”: 0
},
{
“ID”: 1
}
]
}
]
}
}


#### Human Readable Output

##### Kafka Topics

| **Name** | **Partitions** |
| --- | --- |
| test-topic1 | {'ID': 0} |
| test-topic2 | {'ID': 0, 'EarliestOffset': 0, 'OldestOffset': 3}, {'ID': 1, 'EarliestOffset': 0, 'OldestOffset': 4} |

### kafka-publish-msg

***
Publishes a message to Kafka.

#### Base Command

`kafka-publish-msg`

#### Input

| **Argument Name** | **Description** | **Required** |
| --- | --- | --- |
| topic | A topic to publish messages to. | Required |
| value | Message value (string). | Required |
| partitioning_key | Message partition (number). | Optional |
| value_schema_type | Value schema type. If not set, no schema is used, and schema-related arguments are ignored. Possible values are: AVRO. | Optional |
| value_schema_str | Value schema in string format. Used only if `schema_type` is provided and `schema_subject_name` is not. Mutually exclusive with `schema_subject_name`. | Optional |
| value_schema_subject_name | Value schema subject name to retrieve the latest schema version from the registry. Used only if `schema_type` is provided and `schema_str` is not. Mutually exclusive with `schema_str`. | Optional |

#### Context Output

There is no context output for this command.

#### Command Example

```!kafka-publish-msg topic=test-topic value="test message"```

#### Human Readable Output

Message was successfully produced to topic 'test-topic', partition 0

### kafka-consume-msg

***
Consumes a single Kafka message.

#### Base Command

`kafka-consume-msg`

#### Input

| **Argument Name** | **Description** | **Required** |
| --- | --- | --- |
| topic | A topic to get messages from. | Required |
| offset | Message offset to filter by. Acceptable values are 'Earliest', 'Latest', or any other offest number. Default is Earliest. | Optional |
| partition | Partition (number). | Optional |
| poll_timeout | Poll timeout to consume the message. | Optional |

#### Context Output

| **Path** | **Type** | **Description** |
| --- | --- | --- |
| Kafka.Topic.Name | string | Name of the topic. |
| Kafka.Topic.Message.Value | string | Value of the message. |
| Kafka.Topic.Message.Offset | number | Offset of the value in the topic. |

#### Command Example

```!kafka-consume-msg topic=test-topic offset=latest```

#### Context Example

{
“Kafka”: {
“Topic”: {
“Message”: {
“Value”: “test message”,
“Offset”: 11
},
“Name”: “test-topic”
}
}
}


#### Human Readable Output

##### Message consumed from topic 'test'

| **Offset** | **Message** |
| --- | --- |
| 11 | test message |

### kafka-fetch-partitions

***
Fetches partitions for a topic.

#### Base Command

`kafka-fetch-partitions`

#### Input

| **Argument Name** | **Description** | **Required** |
| --- | --- | --- |
| topic | A topic to fetch partitions for. | Required |

#### Context Output

| **Path** | **Type** | **Description** |
| --- | --- | --- |
| Kafka.Topic.Name | string | Name of topic. |
| Kafka.Topic.Partition | number | Prints all partitions for a topic. |

#### Command Example

```!kafka-fetch-partitions topic=test```

### Context Example

{
“Kafka”: {
“Topic”: {
“Partition”: [
0,
1,
2
],
“Name”: “test”
}
}
}
```

Human Readable Output

Available partitions for topic ‘test’

Partitions
0
1
2

Configuration of SASL_SSL PLAIN

Make sure you have the broker port which supports SSL connection.
Add ‘broker_address:port’ to the brokers list.
Provide the CA root certificate in the ‘CA certificate of Kafka server (.cer)’ section.
If your client certificate is password protected, provide the password in the ‘Client certificate key password (if required)’ section.
Provide SASL PLAIN Username and SASL PLAIN Password

Note: SASL is supported only when used in combination with SSL.

Important:
This integration also supports users with consumer only permissions.

Configuration parameters

brokers — CSV list of Kafka brokers to connect to, e.g., 172.16.20.207:9092,172.16.20.234:9093 (required)
group_id — Consumer group ID
use_ssl — Use TLS for connection
use_sasl — Use SASL PLAIN for connection (using SSL)
insecure — Trust any certificate (not secure)
ca_cert — CA certificate of Kafka server (.cer)
client_cert — Client certificate (.cer)
client_cert_key — Client certificate key (.key)
additional_password —
credentials — SASL PLAIN Username
topic — Topic to fetch incidents from (Required for fetch incidents)
partition — CSV list of partitions to fetch messages from
first_fetch — Offset to fetch messages from (Exclusive)
max_fetch — Maximum number of messages to fetch
stop_consuming_upon_timeout — Stop consuming upon timeout
isFetch — Fetch incidents
incidentType — Incident type
incidentFetchInterval — Incidents Fetch Interval
max_bytes_per_message — Maximum number of bytes per message
schema_registry_url — Schema Registry URL
schema_registry_credentials — Schema Registry Username

Commands (4)

kafka-consume-msg
Consumes a single Kafka message.
kafka-fetch-partitions
Fetch partitions for a topic.
kafka-print-topics
Prints all topics and their partitions.
kafka-publish-msg
Publishes a message to Kafka.

Source

KafkaV3.py KafkaV3_test.py KafkaV3.yml KafkaV3_description.md

Download History Raw

category: Messaging and Conferencing
provider: Open Source
commonfields:
  id: KafkaV3
  version: -1
configuration:
- display: CSV list of Kafka brokers to connect to, e.g., 172.16.20.207:9092,172.16.20.234:9093
  name: brokers
  required: true
  type: 0
  section: Connect
- display: Consumer group ID
  additionalinfo: This group ID will be used when fetching incidents and preforming consumer commands. If not set the group id 'xsoar_group' will be used.
  name: group_id
  type: 0
  required: false
  section: Collect
- display: Use TLS for connection
  name: use_ssl
  type: 8
  required: false
  section: Connect
- display: Use SASL PLAIN for connection (using SSL)
  name: use_sasl
  type: 8
  section: Connect
- display: Trust any certificate (not secure)
  name: insecure
  type: 8
  required: false
  section: Connect
- display: CA certificate of Kafka server (.cer)
  name: ca_cert
  type: 12
  required: false
  section: Connect
- display: Client certificate (.cer)
  name: client_cert
  type: 12
  required: false
  section: Connect
- display: Client certificate key (.key)
  name: client_cert_key
  type: 14
  required: false
  section: Connect
- displaypassword: Client certificate key password (if required)
  hiddenusername: true
  name: additional_password
  type: 9
  required: false
  section: Connect
- display: SASL PLAIN Username
  name: credentials
  type: 9
  displaypassword: SASL PLAIN Password
  required: false
  section: Connect
- display: Topic to fetch incidents from (Required for fetch incidents)
  name: topic
  type: 0
  required: false
  section: Collect
- display: CSV list of partitions to fetch messages from
  name: partition
  type: 0
  required: false
  section: Collect
- additionalinfo: The initial offset to start fetching from, not including the value set (e.g., if 3 is set, the first event that will be fetched will be from offset 4). If you want to start from the earliest or latest, type in 'earliest' or 'latest'.
  display: Offset to fetch messages from (Exclusive)
  name: first_fetch
  type: 0
  required: false
  section: Collect
- defaultvalue: '50'
  display: Maximum number of messages to fetch
  name: max_fetch
  type: 0
  required: false
  section: Collect
- display: Stop consuming upon timeout
  additionalinfo: When fetching a significant number of messages (100+), it's advisable to halt message consumption upon timeout. This ensures that the fetch terminates if no messages are received after a specified duration, instead of requesting messages until reaching the maximum number of messages to fetch.
  name: stop_consuming_upon_timeout
  type: 8
  advanced: true
  required: false
  section: Collect
- display: Fetch incidents
  name: isFetch
  type: 8
  required: false
  section: Collect
- display: Incident type
  name: incidentType
  type: 13
  required: false
  section: Collect
- display: Incidents Fetch Interval
  name: incidentFetchInterval
  defaultvalue: '1'
  required: false
  type: 19
  section: Collect
  advanced: true
- additionalinfo: |-
    The maximum number of message bytes to retrieve in each attempted fetch request. Should be in multiples of 1024. If the fetching process takes a long time, consider increasing this value. Default is '1048576'.
  defaultvalue: '1048576'
  display: Maximum number of bytes per message
  name: max_bytes_per_message
  type: 0
  required: false
  section: Collect
- display: Schema Registry URL
  name: schema_registry_url
  required: false
  type: 0
  section: Connect
- display: Schema Registry Username
  displaypassword: Schema Registry Password
  name: schema_registry_credentials
  required: false
  type: 9
  section: Connect
description: Kafka is an open source distributed streaming platform.
display: Kafka v3
name: KafkaV3
script:
  commands:
  - arguments:
    - auto: PREDEFINED
      defaultValue: 'true'
      description: Whether to include the first and last offset for a topic, when printing a list of topics and partitions.
      name: include_offsets
      predefined:
      - 'true'
      - 'false'
    description: Prints all topics and their partitions.
    name: kafka-print-topics
    outputs:
    - contextPath: Kafka.Topic.Name
      description: Kafka topic name.
      type: String
    - contextPath: Kafka.Topic.Partitions.ID
      description: Topic partition ID.
      type: Number
    - contextPath: Kafka.Topic.Partitions.EarliestOffset
      description: Topic partition earliest offset.
      type: Number
    - contextPath: Kafka.Topic.Partitions.LatestOffset
      description: Topic partition latest offset.
      type: Number
  - arguments:
    - description: A topic to publish messages to.
      name: topic
      required: true
    - description: Message value (string).
      name: value
      required: true
    - description: Message partition (number).
      name: partitioning_key
    - auto: PREDEFINED
      description: Value schema type. If not set, no schema is used, and schema-related arguments are ignored.
      name: value_schema_type
      predefined:
      - AVRO
    - description: Value schema in string format. Used only if `schema_type` is provided and `schema_subject_name` is not. Mutually exclusive with `schema_subject_name`.
      name: value_schema_str
    - description: Value schema subject name to retrieve the latest schema version from the registry. Used only if `schema_type` is provided and `schema_str` is not. Mutually exclusive with `schema_str`.
      name: value_schema_subject_name
    description: Publishes a message to Kafka.
    name: kafka-publish-msg
  - arguments:
    - description: A topic to get messages from.
      name: topic
      required: true
    - defaultValue: Earliest
      description: Message offset to filter by. Acceptable values are 'Earliest', 'Latest', or an offest number.
      name: offset
    - description: Partition (number).
      name: partition
    - description: Poll timeout to consume a message.
      name: poll_timeout
      defaultValue: 1.0
    description: Consumes a single Kafka message.
    name: kafka-consume-msg
    outputs:
    - contextPath: Kafka.Topic.Name
      description: Name of the topic.
      type: string
    - contextPath: Kafka.Topic.Message.Value
      description: Value of the message.
      type: string
    - contextPath: Kafka.Topic.Message.Offset
      description: Offset of the value in the topic.
      type: number
  - arguments:
    - description: A topic to fetch partitions for.
      name: topic
      required: true
    description: Fetch partitions for a topic.
    name: kafka-fetch-partitions
    outputs:
    - contextPath: Kafka.Topic.Name
      description: Name of topic.
      type: string
    - contextPath: Kafka.Topic.Partition
      description: Prints all partitions for a topic.
      type: number
  isfetch: true
  runonce: false
  script: '-'
  type: python
  subtype: python3
  dockerimage: demisto/vendors-sdk:1.0.0.10120494
fromversion: 6.1.0
tests:
- KafkaV3 Test
sectionorder:
- Connect
- Collect