# Configure Kafka Output to send ByteArray

**URL:** <https://discuss.elastic.co/t/configure-kafka-output-to-send-bytearray/60758>\
**Category:** Logstash\
**Created:** [September 16, 2016, 9:35pm UTC](https://discuss.elastic.co/t/configure-kafka-output-to-send-bytearray/60758 "2016-09-16T21:35:35Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![fsellecchia](https://avatars.discourse-cdn.com/v4/letter/f/f17d59/32.png) [@fsellecchia](https://discuss.elastic.co/u/fsellecchia)\
**Post date:** [September 16, 2016, 9:35pm UTC](https://discuss.elastic.co/t/configure-kafka-output-to-send-bytearray/60758/1 "2016-09-16T21:35:35Z")

</div>

I'm working with logstash version 2.3.4 and trying to write an Array of Bytes to kafka.  
For doing this, I built a filter that encodes a message in avro. All the enconding / deconding unit tests on the filter work fine.  
I need to send the Array of Bytes to Kafka, heres is the logstash.conf file:

input {  
file {  
add\_field =\> ["timestamp", ""]  
path =\> "/var/log/httpd/access\_log._"  
start\_position =\> end  
type =\> "myAvroEvent"  
sincedb\_path =\> "/root/.sincedb_"  
sincedb\_write\_interval =\> 15  
}

}

filter {  
if [type] == "myAvroEvent" {  
myAvroEvent{  
message =\> message  
}  
}  
}

output {

```
    if [type] == "myAvroEvent" {
            kafka {
                    bootstrap_servers => "localhost:9092"
                    topic_id => "my_topic"
                    metadata_max_age_ms => "1000"
                    value_serializer => "org.apache.kafka.common.serialization.ByteArraySerializer"
            }
    }

```

}

When serializing to kafka logstash.log shows this:

{:timestamp=\>"2016-09-16T21:08:41.741000+0000", :message=\>"kafka producer threw exception, restarting", :exception=\>org.apache.kafka.common.errors.SerializationException: Can't convert value of class java.lang.String to class org.apache.kafka.common.serialization.ByteArraySerializer specified in value.serializer, :level=\>:warn}

I tried to use plain codec as well and I get the same error.

Any ideas about what migth be wrong? Is logstash.conf file configured properly?

Thanks in advance.  
Fernando.

---

<div class="post-metadata">

**Author:** ![fsellecchia](https://avatars.discourse-cdn.com/v4/letter/f/f17d59/32.png) [@fsellecchia](https://discuss.elastic.co/u/fsellecchia)\
**Post date:** [September 21, 2016, 7:44pm UTC](https://discuss.elastic.co/t/configure-kafka-output-to-send-bytearray/60758/2 "2016-09-21T19:44:00Z")

</div>

OK, finally I found a solution for this issue.  
Kafka output plugin uses by default "plain" codec. This codec converts the message into a string before sending it to kafka. So the filter was sending a byte array, but the codec convert it to a string and the plugin tries to serialize the message using ByteArraySerializer.  
The way to solve it is to set codec that sends the message as it is without converting it. Since I didn't find a codec for doing that I created a new one.

# encoding: utf-8  
require "logstash/codecs/base"

# The "none" codec send the message as it is without transforming it.

# 

class LogStash::Codecs::NoCodec \< LogStash::Codecs::Base  
config\_name "nocodec"

public  
def register  
end

public  
def decode(data)  
yield LogStash::Event.new("message" =\> data)  
end # def decode

public  
def encode(event)  
@on\_event.call(event, event["message"])  
end # def encode

end # class LogStash::Codecs::NoCodec

When configuring kafka output on logstash.conf, just add this line  
codec =\> nocodec

Thanks

---

<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:** [July 6, 2017, 4:37am UTC](https://discuss.elastic.co/t/configure-kafka-output-to-send-bytearray/60758/3 "2017-07-06T04:37:31Z")

</div>