# Kafka -\> Logstash -\> ES ..... using avro\_schema\_registry plugin & decorate\_events & mutate = Fail. I would like to get Key info into ES using Mutate

**URL:** <https://discuss.elastic.co/t/kafka-logstash-es-using-avro-schema-registry-plugin-decorate-events-mutate-fail-i-would-like-to-get-key-info-into-es-using-mutate/146916>\
**Category:** Logstash\
**Created:** [August 31, 2018, 8:56pm UTC](https://discuss.elastic.co/t/kafka-logstash-es-using-avro-schema-registry-plugin-decorate-events-mutate-fail-i-would-like-to-get-key-info-into-es-using-mutate/146916 "2018-08-31T20:56:22Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![Robert\_Duffy](https://avatars.discourse-cdn.com/v4/letter/r/41988e/32.png) [@Robert\_Duffy](https://discuss.elastic.co/u/Robert_Duffy)\
**Post date:** [August 31, 2018, 8:56pm UTC](https://discuss.elastic.co/t/kafka-logstash-es-using-avro-schema-registry-plugin-decorate-events-mutate-fail-i-would-like-to-get-key-info-into-es-using-mutate/146916/1 "2018-08-31T20:56:22Z")

</div>

I am not having success with the following:

Kafka Input (both key & body avro schema are in schema registry)

I would like to pull fields from they Key and combine with the body and output to ES.

decorate\_events=true, with the codec=\>avro\_schema\_registry plugin, w/o the key\_deserializer class allows mutate operation to find values **however key is in Bytes**.

decorate\_events=true with they key\_deserializer class defined fails w/ exception

input {  
.......  
...  
#decorate\_events =\> true  
codec =\> avro\_schema\_registry {  
endpoint =\> "http://...........:8081"  
}  
value\_deserializer\_class =\> "org.apache.kafka.common.serialization.ByteArrayDeserializer"  
key\_deserializer\_class =\> "org.apache.kafka.common.serialization.ByteArrayDeserializer"  
}

filter {  
mutate {  
add\_field =\> {  
"my\_kafka\_partition" =\> "%{[@metadata][kafka][partition]}"  
"my\_kafka\_offset" =\> "%{[@metadata][kafka][offset]}"  
"my\_kafka\_key" =\> "%{message}"  
}  
}  
}

```
exception when using decorate_events=> true & the key_deserializer_class defined

Exception in thread "Ruby-0-Thread-39: :1" org.logstash.MissingConverterException: Missing Converter handling for full class name=[B, simple name=byte[]
        at org.logstash.Valuefier.fallbackConvert(Valuefier.java:97)
        at org.logstash.Valuefier.convert(Valuefier.java:75)
        at org.logstash.Valuefier.lambda$static$3(Valuefier.java:51)
        at org.logstash.Valuefier.convert(Valuefier.java:73)
        at org.logstash.ext.JrubyEventExtLibrary$RubyEvent.ruby_set_field(JrubyEventExtLibrary.java:99)
        at org.logstash.ext.JrubyEventExtLibrary$RubyEvent$INVOKER$i$2$0$ruby_set_field.call(JrubyEventExtLibrary$RubyEvent$INVOKER$i$2$0$ruby_set_field.gen)
        at org.jruby.internal.runtime.methods.JavaMethod$JavaMethodN.call(JavaMethod.java:741)
        at org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:145)
        at home.rduffy.logstash.logstash_minus_6_dot_4_dot_0.vendor.bundle.jruby.$2_dot_3_dot_0.gems.logstash_minus_input_minus_kafka_minus_8_dot_1_dot_1.lib.logstash.inputs.kafka.RUBY$block$thread_runner$3(/home/rduffy/logstash/logstash-6.4.0/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.1.1/lib/logstash/inputs/kafka.rb:263)
        at org.jruby.runtime.CompiledIRBlockBody.yieldDirect(CompiledIRBlockBody.java:156)
        at org.jruby.runtime.BlockBody.yield(BlockBody.java:114)
        at org.jruby.runtime.Block.yield(Block.java:165)
        at org.jruby.ir.runtime.IRRuntimeHelpers.yield(IRRuntimeHelpers.java:415)
        at org.jruby.ir.targets.YieldSite.yield(YieldSite.java:87)
        at home.rduffy.logstash.logstash_minus_6_dot_4_dot_0.vendor.bundle.jruby.$2_dot_3_dot_0.gems.logstash_minus_codec_minus_avro_schema_registry_minus_1_dot_1_dot_0.lib.logstash.codecs.avro_schema_registry.RUBY$method$decode$0(/home/rduffy/logstash/logstash-6.4.0/vendor/bundle/jruby/2.3.0/gems/logstash-codec-avro_schema_registry-1.1.0/lib/logstash/codecs/avro_schema_registry.rb:222)
        at home.rduffy.logstash.logstash_minus_6_dot_4_dot_0.vendor.bundle.jruby.$2_dot_3_dot_0.gems.logstash_minus_codec_minus_avro_schema_registry_minus_1_dot_1_dot_0.lib.logstash.codecs.avro_schema_registry.RUBY$method$decode$0$ __VARARGS__ (/home/rduffy/logstash/logstash-6.4.0/vendor/bundle/jruby/2.3.0/gems/logstash-codec-avro_schema_registry-1.1.0/lib/logstash/codecs/avro_schema_registry.rb)
        at org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:77)
        at org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:93)
        at org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:145)
        at home.rduffy.logstash.logstash_minus_6_dot_4_dot_0.vendor.bundle.jruby.$2_dot_3_dot_0.gems.logstash_minus_input_minus_kafka_minus_8_dot_1_dot_1.lib.logstash.inputs.kafka.RUBY$block$thread_runner$2(/home/rduffy/logstash/logstash-6.4.0/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.1.1/lib/logstash/inputs/kafka.rb:256)
        at org.jruby.runtime.CompiledIRBlockBody.yieldDirect(CompiledIRBlockBody.java:156)
        at org.jruby.runtime.BlockBody.yield(BlockBody.java:114)
        at org.jruby.runtime.Block.yield(Block.java:165)
        at org.jruby.javasupport.ext.JavaLang$Iterable.each(JavaLang.java:93)
        at org.jruby.javasupport.ext.JavaLang$Iterable$INVOKER$s$0$0$each.call(JavaLang$Iterable$INVOKER$s$0$0$each.gen)
        at org.jruby.internal.runtime.methods.JavaMethod$JavaMethodZeroBlock.call(JavaMethod.java:498)
        at org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:145)
        at home.rduffy.logstash.logstash_minus_6_dot_4_dot_0.vendor.bundle.jruby.$2_dot_3_dot_0.gems.logstash_minus_input_minus_kafka_minus_8_dot_1_dot_1.lib.logstash.inputs.kafka.RUBY$block$thread_runner$1(/home/rduffy/logstash/logstash-6.4.0/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.1.1/lib/logstash/inputs/kafka.rb:255)
        at org.jruby.runtime.CompiledIRBlockBody.callDirect(CompiledIRBlockBody.java:145)
        at org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:71)
        at org.jruby.runtime.Block.call(Block.java:124)
        at org.jruby.RubyProc.call(RubyProc.java:289)
        at org.jruby.RubyProc.call(RubyProc.java:246)
```

---

<div class="post-metadata">

**Author:** ![Robert\_Duffy](https://avatars.discourse-cdn.com/v4/letter/r/41988e/32.png) [@Robert\_Duffy](https://discuss.elastic.co/u/Robert_Duffy)\
**Post date:** [September 20, 2018, 10:27pm UTC](https://discuss.elastic.co/t/kafka-logstash-es-using-avro-schema-registry-plugin-decorate-events-mutate-fail-i-would-like-to-get-key-info-into-es-using-mutate/146916/2 "2018-09-20T22:27:17Z")

</div>

ping ping... anyone?

---

<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:** [October 18, 2018, 10:27pm UTC](https://discuss.elastic.co/t/kafka-logstash-es-using-avro-schema-registry-plugin-decorate-events-mutate-fail-i-would-like-to-get-key-info-into-es-using-mutate/146916/3 "2018-10-18T22:27:19Z")

</div>

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