KafkaV3

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

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

Context Output

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’

Configuration of SASL_SSL PLAIN

1. Make sure you have the broker port which supports SSL connection.
2. Add ‘broker_address:port’ to the brokers list.
3. Provide the CA root certificate in the ‘CA certificate of Kafka server (.cer)’ section.
4. If your client certificate is password protected, provide the password in the ‘Client certificate key password (if required)’ section.
5. 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.