fix(inputs.kafka_consumer): Use per-message parser to avoid races

[toanju]

Using parsers like json_v2 will result in undesired parser results. This switches to the ParserFunc pattern to create a dedicated parser in each thread.

Required for all PRs

• Updated associated README.md.
• Wrote appropriate unit tests.
• Pull request title or commits are in conventional commit format

fixes #13888

[powersj]

Hi,

Using parsers like json_v2 will result in undesired parser results.

Can you please file an issue and document what you were seeing please?

[toanju]

related to #13888

let me know if you need more info

[srebhan]

While that might work it will strictly serialize the parsing... Should we instead change the plugin to the parser-func interface and create a new parser each time (or use a sync.Pool of parsers)?

My concern is that this will hamper performance on larger messages or slow parsers...

[toanju]

Performance is not important if you get wrong results. Though, I agree that the mutex in this place may have an impact on other parsers and the JSON parser might be one that folks to not use a lot.

I am open to implement this, however, currently I'd need to look into some of the suggested options to get a better understanding. Hence, if someone can point into the right direction this would be awesome.

[toanju]

@srebhan, implemented the ParserFunc for now. Looking forward to additional feedback.

[telegraf-tiger]

Download PR build artifacts for linux_amd64.tar.gz, darwin_amd64.tar.gz, and windows_amd64.zip.
Downloads for additional architectures and packages are available below.

👍 This pull request doesn't change the Telegraf binary size

📦 Click here to get additional PR build artifacts

Artifact URLs

DEB	RPM	TAR GZ	ZIP
amd64.deb	aarch64.rpm	darwin_amd64.tar.gz	windows_amd64.zip
arm64.deb	armel.rpm	darwin_arm64.tar.gz	windows_arm64.zip
armel.deb	armv6hl.rpm	freebsd_amd64.tar.gz	windows_i386.zip
armhf.deb	i386.rpm	freebsd_armv7.tar.gz
i386.deb	ppc64le.rpm	freebsd_i386.tar.gz
mips.deb	riscv64.rpm	linux_amd64.tar.gz
mipsel.deb	s390x.rpm	linux_arm64.tar.gz
ppc64el.deb	x86_64.rpm	linux_armel.tar.gz
riscv64.deb		linux_armhf.tar.gz
s390x.deb		linux_i386.tar.gz
		linux_mips.tar.gz
		linux_mipsel.tar.gz
		linux_ppc64le.tar.gz
		linux_riscv64.tar.gz
		linux_s390x.tar.gz

[srebhan]

Thank you for the quick update and your contribution in general @toanju!

[toanju]

Many thanks @srebhan and @powersj 🎉

[athornton]

This makes using a schema registry impossible, because the schema registry cache is down in the Avro parser, and a new parser means an always-empty cache, which means we query the schema registry once per measurement. Since in normal operations there are many, many more measurements than schemas, this creates an enormous load on the schema registry, and it also means that parsing each measurement takes much too long, since it has to wait on a network call.

#14025 is what I'm working on to address this. @srebhan suggested an approach based on adding Clone() to the parser layer, and I'm trying to work through that; this looks to me like I have to understand each parser, whether or not it's thread-safe, and implement the correct Clone() method for each parser. That's a daunting task. Another possibility might be to associate the schema registry with the Kafka input layer rather than the Avro parser layer, since you need to be using both Kafka and Avro (I think) to use a Schema registry.