fix: reject mismatched decimals from avro topics

[jzaralim]

Description

Fixes #7170

If you run the following statements

CREATE STREAM A (dec DECIMAL(4,2)) WITH (topic='topic', value_format='AVRO');
CREATE STREAM B (dec DECIMAL(6,3)) WITH (topic='topic', value_format='AVRO');
CREATE STREAM C AS SELECT * FROM A;
INSERT INTO B VALUES (123.456);

You would end up with a org.apache.kafka.common.errors.SerializationException in the logs, instead of rejecting the value to begin with.

And querying A would return the mismatched decimal

SELECT * FROM A EMIT CHANGES;
+---------+
|DEC      |
+---------+
|123.456  |

This PR changes the behaviour so that if there is a mismatching decimal, it first tries to convert the topic decimal to the ksql schema, and if that is impossible, then it skips that value and the logs a org.apache.kafka.common.errors.SerializationException in the streams log.

Testing done

Manually tested, added unit test

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[abbccdda]

Maybe a dumb question, could we support a conversion from 1234.5 to a 5 precision and 0 scale decimal number? Are there any other invalid cases we could test out?

[jzaralim]

No, we don't support rounding for conversions now. Added a test for that case.

[abbccdda]

How did we reach the assumption that given type of BYTES would require a decimal conversion?

[jzaralim]

Decimals in Avro are represented as bytes. This function is called earlier on to reject any BYTES that don't represent decimals. Right now, we don't have a BYTES type in ksqlDB so all incoming bytes must be decimals, but this logic will change once we implement it.

[abbccdda]

LGTM, thanks for the PR! @jzaralim