# Kafka-Elasticsearch Logstash Configuration Error

**URL:** <https://discuss.elastic.co/t/kafka-elasticsearch-logstash-configuration-error/330130>\
**Category:** Logstash\
**Tags:** docker\
**Created:** [April 17, 2023, 12:42pm UTC](https://discuss.elastic.co/t/kafka-elasticsearch-logstash-configuration-error/330130 "2023-04-17T12:42:00Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![Mustafa\_AYDOGDU](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mustafa_aydogdu/32/119908_2.png) [@Mustafa\_AYDOGDU](https://discuss.elastic.co/u/Mustafa_AYDOGDU)\
**Post date:** [April 17, 2023, 12:42pm UTC](https://discuss.elastic.co/t/kafka-elasticsearch-logstash-configuration-error/330130/1 "2023-04-17T12:42:01Z")

</div>

I have a logstash pipeline which gets data from kafka and sends it to elasticsearch. However, in elasticsearch, data is not represented correctly.  
In this data I want it to be just field:value. But it is field:[value,field]. I could not figure out the reason.

 ![image](https://us1.discourse-cdn.com/elastic/original/3X/e/6/e699772f8785207f676582e1a8d8ca56e0adae93.png)  
Here is my logstash conf file:

```auto
input {
    kafka {
        codec => "json"
        bootstrap_servers => "kafka:9092"
        topics => ["tweet_user_id", "detailed_users", "detailed_tweets"]
        decorate_events => true
    }
}

filter {
    if [@metadata][kafka][topic] == "detailed_users" {
        mutate {
            add_field => { 
                "user_id" => "%{[message][user_id]}"
            }
            add_field => {
                "user_name" => "%{[message][user_name]}"
            }
            add_field => {
                "bio" => "%{[message][bio]}"
            }
            add_field => {
                "location" => "%{[message][location]}"
            }
            add_field => {
                "website" => "%{[message][website]}"
            }
            add_field => {
                "join_date" => "%{[message][join_date]}"
            }
            add_field => {
                "tweet_count" => "%{[message][tweet_count]}"
            }
            add_field => {
                "following_count" => "%{[message][following_count]}"
            }
            add_field => {
                "followers_count" => "%{[message][followers_count]}"
            }
            add_field => {
                "like_count" => "%{[message][like_count]}"
            }
            remove_field => ["message", "event"]
        }
    }
}

output {
    if [@metadata][kafka][topic] == "tweet_user_id" {
        elasticsearch {
            hosts => ["elasticsearch:9200"]
            index => "tweet_user_id_index"
        }
    }

    else if [@metadata][kafka][topic] == "detailed_users" {
        elasticsearch {
            hosts => ["elasticsearch:9200"]
            index => "detailed_users_index"
        }
    }

    else if [@metadata][kafka][topic] == "detailed_tweets" {
        elasticsearch {
            hosts => ["elasticsearch:9200"]
            index => "detailed_tweets_index"
        }
    }
    stdout { 
        codec => json_lines 
    }
}

```

Thanks in advance.

---

<div class="post-metadata">

**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\
**Post date:** [April 17, 2023, 2:52pm UTC](https://discuss.elastic.co/t/kafka-elasticsearch-logstash-configuration-error/330130/2 "2023-04-17T14:52:20Z")

</div>

The json codec is creating fields at the top level. You are then trying to move them from inside the [message] field using mutate+add\_field, but since the source fields do not exist, that just converts all of the fields to arrays. Remove all of the add\_field mutates.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [May 15, 2023, 2:52pm UTC](https://discuss.elastic.co/t/kafka-elasticsearch-logstash-configuration-error/330130/3 "2023-05-15T14:52:32Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.