# Kafka input convert to json
**URL:** https://discuss.elastic.co/t/kafka-input-convert-to-json/215785
**Category:** Logstash
**Created:** [January 20, 2020, 6:22pm UTC](https://discuss.elastic.co/t/kafka-input-convert-to-json/215785 "2020-01-20T18:22:13Z")
**Posts on this page:** 7
**Page:** 1
### Author:  [@elasticforme](https://discuss.elastic.co/u/elasticforme)
#### Post date: [January 20, 2020, 6:22pm UTC](https://discuss.elastic.co/t/kafka-input-convert-to-json/215785/1 "2020-01-20T18:22:13Z")
I have use input as kafka and my message is like
2020-01-20 10:47:48: partition=0 offset=187965 machine=host1 queue=qhost1 comman=1298316293 action=REMOVE
I want this to be separated out. how do I use filter
```
input {
kafka {
bootstrap_servers => "kaf001:9092"
topics => "events.commands"
}
}
filter {}
output {
stdout { codec => rubydubug }
}
```
I want output like
```
{
mydate: 2020-01-20 10:47:48
partition: 0
offset:187965
machine: host1
queue: qhost1
command: 1298316293
action: REMOVE
}
```
but output has some kind of special character. and Looks like this at this time.
"message" =\> "\u0001\u000E\u0010qhost1����\t\fREMOVE\u0002�����[\u0000"
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [January 20, 2020, 9:22pm UTC](https://discuss.elastic.co/t/kafka-input-convert-to-json/215785/2 "2020-01-20T21:22:30Z")
That looks to me as though you have the wrong encoding. You can specify this using the [charset](https://www.elastic.co/guide/en/logstash/7.5/plugins-codecs-plain.html#plugins-codecs-plain-charset) option on the codec for the input. That is, add something like
```
codec => plain { charset => "UTF-8" }
```
I do not think that text is UTF-8, I am just giving you an example.
Once your messages look good I would use dissect to take the prefix off the line and then a kv filter, similar to [this](https://discuss.elastic.co/t/parsing-firewall-logs-in-logstash/212786/2).
---
### Author:  [@elasticforme](https://discuss.elastic.co/u/elasticforme)
#### Post date: [January 21, 2020, 4:06pm UTC](https://discuss.elastic.co/t/kafka-input-convert-to-json/215785/3 "2020-01-21T16:06:20Z")
ohh found out that they are using avro to put message in kafka and got avro schema file from them but now getting different error.
```
Error: negative length -1 given
Exception: ArgumentError
Stack: org/jruby/ext/stringio/StringIO.java:852:in `read'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:106:in `read'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:93:in `read_bytes'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:99:in `read_string'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:276:in `read_data'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:363:in `block in read_record'
org/jruby/RubyArray.java:1800:in `each'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:361:in `read_record'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:287:in `read_data'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:252:in `read'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-codec-avro-3.2.3-java/lib/logstash/codecs/avro.rb:77:in `decode'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.0.0-java/lib/logstash/inputs/kafka.rb:258:in `bk in thread_runner'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.0.0-java/lib/logstash/inputs/kafka.rb:257:in `bk in thread_runner'
[FATAL] 2020-01-21 09:28:35.646 [LogStash::Runner] runner - An unexpected error occurred! {:error=>#
```
---
### Author:  [@elasticforme](https://discuss.elastic.co/u/elasticforme)
#### Post date: [January 22, 2020, 2:05pm UTC](https://discuss.elastic.co/t/kafka-input-convert-to-json/215785/4 "2020-01-22T14:05:27Z")
found more that they are encoding data. how do I decode it?
---
### Author:  [@elasticforme](https://discuss.elastic.co/u/elasticforme)
#### Post date: [January 22, 2020, 3:56pm UTC](https://discuss.elastic.co/t/kafka-input-convert-to-json/215785/5 "2020-01-22T15:56:56Z")
Team who is putting this message in log telling me to remove first character.
it does not make sense to me but i have to test and prove them
how do I remove very first character from message that I get ?
---
### Author:  [@elasticforme](https://discuss.elastic.co/u/elasticforme)
#### Post date: [January 22, 2020, 4:23pm UTC](https://discuss.elastic.co/t/kafka-input-convert-to-json/215785/6 "2020-01-22T16:23:20Z")
I use this, ignore last request
mutate { gsub =\> ["message", "^.", ""] }
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [February 19, 2020, 4:23pm UTC](https://discuss.elastic.co/t/kafka-input-convert-to-json/215785/7 "2020-02-19T16:23:29Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.