[NopeCharon]

Fixes #619

kafka2 and rdkafka2 read message_key_key, partition_key_key (kafka2 only) and partition_key with record[name], so the message key, the partition key and the partition had to be top-level fields. A value in a nested field, such as kubernetes.pod_name, first had to be copied to the top level with record_transformer. This was asked in #337. headers_from_record, exclude_fields and record_key of the same outputs already accept the record_accessor syntax.

This change creates a record accessor for each of these parameters in configure and uses it in write:

<match app.**>
  @type kafka2
  message_key_key $.kubernetes.pod_name
  exclude_message_key true
  ...
</match>

• A name that doesn't start with $. or $[ is still a top-level field, even if it contains a dot (kafka.message_key reads the field named kafka.message_key). For Hash records, existing configurations using these top-level field names behave the same.
• exclude_message_key, exclude_partition_key and exclude_partition remove only the nested field. The parent Hash stays in the payload.
• If the nested field is not found, default_message_key, default_partition_key or default_partition is used, as for a missing top-level field. This includes a value on the path that is not a Hash or an Array, such as {"kubernetes": "x"}. There, the accessor's lookup raises TypeError, which would skip the event. Its delete calls String#delete on "x", so with an exclude_* parameter, whatever was left of the String would become the key or the partition. The new record_value helper treats both cases as a missing field and leaves the record unchanged.

topic_key is not changed. The topic comes from the buffer chunk keys, not from the record.

README: one paragraph in "Load balancing" with an example and the rules above, and one sentence in the rdkafka2 section.

Tests (test_out_kafka2.rb and test_out_rdkafka2.rb, with a stub producer, no broker):

• nested message key, partition key and partition, with dot and bracket notation
• the exclude_* parameters remove only the nested fields
• defaults are used for a missing parent, a missing field, and a parent that is a String or an Array, with and without exclude_*
• a dotted name without $. is still a top-level field

I ran both test files without the broker-dependent WriteTest cases on Ruby 3.4 with fluentd 1.20.1, ruby-kafka 1.5.0 and rdkafka 0.31.0. All pass. With lib/ reverted, the nested-key and nested exclude tests fail in both files.

[Watson1978]

Thanks for the PR. Nested keys work as described for Hash records. I compared master and this branch with many existing configurations: top-level names, values such as nil, false, numbers, Hash and Array, with and without exclude_*. Both sent the same keys, partitions and payloads.

There is one behavior change for Array and Integer records, and one missing test for the rewritten exclude_* path. Please see the inline comments.

[Watson1978]

MUST: This rescue TypeError also catches the error from an Array or Integer record, not only from a nested lookup.

In master, a record such as [1, 2] or 5 raises TypeError at record[@partition_key_key]. The event is skipped with the "Skip broken event" warning. With this change, record_value returns nil, and the record is sent to Kafka with the default keys and no warning. This happens with the default configuration. For example, in_kafka with the default format json emits a JSON array message as an Array record.

A top-level lookup on a Hash doesn't raise TypeError, so re-raising for other records restores the master behavior for these records:

Suggested change

	rescue TypeError
	rescue TypeError
	raise unless record.is_a?(Hash)

out_rdkafka2.rb needs the same change. If all of exclude_partition_key, exclude_partition and exclude_message_key are set, master happened to send an Array record, because Array#delete doesn't raise. Skipping it in that case too is fine.

This test fails now. It passes with the change, and with the lib/ of master:

def test_write_skips_records_that_are_not_a_hash
  produced = write_records(config, [[1, 2], 5, {"a" => "b"}])

  assert_equal [{record: {"a" => "b"}, key: nil, partition_key: nil, partition: nil}], produced
end

In test_out_rdkafka2.rb, the expected value is [{payload: {"a" => "b"}, key: nil, partition: nil}].

[NopeCharon]

Good catch. It now re-raises unless the record is a Hash, and I added your test. ab8a5cb

[Watson1978]

MUST: Same as the comment on out_kafka2.rb. In master, an Array or Integer record such as [1, 2] or 5 is skipped with a warning. Here, it is sent with the default keys.

Suggested change

	rescue TypeError
	rescue TypeError
	raise unless record.is_a?(Hash)

[NopeCharon]

Same fix here, now in the shared helper. ab8a5cb

[Watson1978]

SHOULD: The tests use exclude_* only with nested paths. But record_value now also handles exclude_* for top-level fields, which existing configurations use. If record_value stops deleting top-level fields, both test files still pass (without the broker-dependent WriteTest cases). I checked this by skipping accessor.delete for accessors that are not nested.

Please add a test with the default field names, for example:

def test_write_excludes_top_level_keys
  conf = config + config_element('ROOT', '', {"exclude_message_key" => true, "exclude_partition_key" => true,
                                              "exclude_partition" => true})
  record = {"message_key" => "mk", "partition_key" => "pk", "partition" => 3, "a" => "b"}

  produced = write_records(conf, [record])

  assert_equal [{record: {"a" => "b"}, key: "mk", partition_key: "pk", partition: 3}], produced
end

And in test_out_rdkafka2.rb:

def test_write_excludes_top_level_keys
  conf = config + config_element('ROOT', '', {"exclude_message_key" => true, "exclude_partition" => true})
  record = {"message_key" => "mk", "partition" => 3, "a" => "b"}

  produced = write_records(conf, [record])

  assert_equal [{payload: {"a" => "b"}, key: "mk", partition: 3}], produced
end

[NopeCharon]

Added test_write_excludes_top_level_keys to both files. ab8a5cb

[Watson1978]

NIT: This method is the same as Fluent::Kafka2Output#record_value. Both outputs already include Fluent::KafkaPluginUtil::PartitionSettings for coerce_partition. You could move record_value to a module in kafka_plugin_util.rb that both outputs include. Then the rescue TypeError fix is made in one place. This is optional.

[NopeCharon]

Moved record_value into PartitionSettings so both outputs share it. ab8a5cb

[Watson1978]

Thanks for the quick update. I checked ab8a5cb:

• With the default configuration, Array and Integer records are skipped again, as in master. test_write_skips_records_that_are_not_a_hash fails without the raise unless record.is_a?(Hash) line.
• test_write_excludes_top_level_keys fails if record_value stops deleting top-level fields.
• Both outputs now share record_value.

I also compared master and this branch again with the same records and configurations as before. Hash records with existing top-level field names get the same keys, partitions and payloads as in master. LGTM.