Getting error while using with record_transformer

[ayZagen]

Hello, I receive error when using source plugin with record_transformer filter.

Here is fluentd config:

<source>
  @type rabbitmq
  tag test
  host  "#{ENV['RABBITMQ_HOST'] || 'rabbitmq'}"
  user  "#{ENV['RABBITMQ_USER'] || 'user'}"
  pass  "#{ENV['RABBITMQ_PASSWORD'] || 'bitnami'}"
  vhost "#{ENV['RABBITMQ_VHOST'] || '/'}"
  queue "#{ENV['RABBITMQ_QUEUE'] || 'test'}"
  <parse>
    @type json
  </parse>
</source>
<filter "**">
  @type record_transformer
  remove_keys $.context.req.headers.accept, $.context.req.headers.accept-language
</filter>

Here is received error (record minified for simplicity):

error="#<NoMethodError: undefined method `merge!' for #<String:0x00007f0139e14bf0>>" location="/usr/local/bundle/gems/fluentd-1.9.2/lib/fluent/plugin/filter_record_transformer.rb:135:in `reform'" tag=test time=1611144715 record="{\"level\":20,\"time\":1611144714919,\"method\":\"GET\",\"status\":200,\"context\":{\"req\":{\"connection\":\"keep-alive\"}}}"

Without record_transformer filter it works. Also I use same filter with fluent-plugin-kafka it works correctly. So, I assume the error is happening because of this plugin.

Thanks