Fix uncaught ParseException when reading Avro from Kafka

[abhishekrb19]

Fixes #13894, #14041.

In StreamChunkParser#parseWithInputFormat, we call byteEntityReader.read() without handling a potential ParseException, which is thrown during this function call by the delegate AvroStreamReader#intermediateRowIterator.
A ParseException can be thrown if an Avro stream has corrupt data or data that doesn't conform to the schema specified or for other decoding reasons. This exception if uncaught, can cause ingestion to fail.

Primary code changes:

• Handle ParseException in the call stack StreamChunkParser.
• Improved the error message when the avro schema id cannot be found.
• Added a new constructor to StreamChunkParser to facilitate unit testing. This allows dependencies to be passed in by the caller, making testing easier.
• Fix miscellaneous typos and javadocs in the modified code.

Unit test changes:

• Add unit tests in StreamChunkParserTest and KafkaIndexTaskTest to test this exception handling behavior alongside setting maximum allowed parse exceptions to different thresholds.
• Replace NoopRowIngestionMeters and a mocked RowIngestionMeters with a more realistic implementation, SimpleRowIngestionMeters, used to validate any parse exceptions.

Release note

Fixes #13894, #14041.

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[abhishekagarwal87]

thank you for fixing this.

[abhishekrb19]

There is the usual flaky kafka slow IT failure unrelated to this change. Hopefully re-running should fix it:

2023-04-28T11:19:34,972 INFO [main] org.apache.druid.security.basic.authorization.db.cache.CoordinatorPollingBasicAuthorizerCacheManager - CoordinatorPollingBasicAuthorizerCacheManager is stopping.
2023-04-28T11:19:34,972 INFO [main] org.apache.druid.security.basic.authorization.db.cache.CoordinatorPollingBasicAuthorizerCacheManager - CoordinatorPollingBasicAuthorizerCacheManager is stopped.
2023-04-28T11:19:34,972 INFO [main] org.apache.druid.security.basic.authentication.db.cache.CoordinatorPollingBasicAuthenticatorCacheManager - CoordinatorPollingBasicAuthenticatorCacheManager is stopping.
2023-04-28T11:19:34,972 INFO [main] org.apache.druid.security.basic.authentication.db.cache.CoordinatorPollingBasicAuthenticatorCacheManager - CoordinatorPollingBasicAuthenticatorCacheManager is stopped.
2023-04-28T11:19:35,014 INFO [Curator-Framework-0] org.apache.curator.framework.imps.CuratorFrameworkImpl - backgroundOperationsLoop exiting
2023-04-28T11:19:35,134 INFO [main] org.apache.zookeeper.ZooKeeper - Session: 0x100000e65e40007 closed
2023-04-28T11:19:35,135 INFO [main-EventThread] org.apache.zookeeper.ClientCnxn - EventThread shut down for session: 0x100000e65e40007
2023-04-28T11:19:35,154 INFO [main] org.apache.druid.java.util.emitter.core.LoggingEmitter - {"feed":"metrics","metric":"jetty/numOpenConnections","service":"","host":"","value":0,"timestamp":"2023-04-28T11:19:35.152Z"}
2023-04-28T11:19:35,154 INFO [main] org.apache.druid.java.util.emitter.core.LoggingEmitter - Close: started [false]
2023-04-28T11:19:35,154 INFO [main] org.apache.druid.java.util.common.lifecycle.Lifecycle - Stopping lifecycle [module] stage [INIT]
Error:  Tests run: 4, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 2,038.216 s <<< FAILURE! - in TestSuite
Error:  testKafkaIndexDataWithLosingCoordinator(org.apache.druid.tests.indexer.ITKafkaIndexingServiceTransactionalSerializedTest)  Time elapsed: 234.292 s
Error:  testKafkaIndexDataWithLosingHistorical(org.apache.druid.tests.indexer.ITKafkaIndexingServiceTransactionalSerializedTest)  Time elapsed: 192.643 s
Error:  testKafkaIndexDataWithLosingOverlord(org.apache.druid.tests.indexer.ITKafkaIndexingServiceTransactionalSerializedTest)  Time elapsed: 189.737 s
Error:  testQueryNilColumnBeforeAndAfterPublishingSegments(org.apache.druid.tests.indexer.ITNilColumnTest)  Time elapsed: 1,344.173 s  <<< FAILURE!
org.apache.druid.java.util.common.ISE: Max number of retries[240] exceeded for Task[Waiting for all tasks to stop]. Failing.
	at org.apache.druid.tests.indexer.ITNilColumnTest.testQueryNilColumnBeforeAndAfterPublishingSegments(ITNilColumnTest.java:126)

[gianm]

The production changes and new tests LGTM. I suggest also adding a test for Kafka indexing itself, to future-proof this bug fix against potential future refactors or entire replacements of StreamChunkParser. That test doesn't need to be Avro-specific.

[abhishekrb19]

@gianm, thanks for the review. Added a UT to Kafka indexing as well.