Rework Flink Schema providers to use existing SerDe and Mutation interfaces #751
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
WalkthroughThis change refactors serialization/deserialization infrastructure, renaming Changes
Sequence Diagram(s)sequenceDiagram
participant FlinkJob
participant FlinkSerDeProvider
participant SerDe
participant DeserializationSchemaBuilder
participant DeserializationSchema
FlinkJob->>FlinkSerDeProvider: build(topicInfo)
FlinkSerDeProvider->>SerDe: (returns SchemaRegistrySerDe or custom SerDe)
FlinkJob->>DeserializationSchemaBuilder: buildSourceIdentityDeserSchema(SerDe, groupBy)
DeserializationSchemaBuilder->>DeserializationSchema: (returns SourceIdentityDeserializationSchema)
FlinkJob->>DeserializationSchema: deserialize(messageBytes, out)
Possibly related PRs
Poem
Tip ⚡️ Faster reviews with caching
Enjoy the performance boost—your workflow just got faster. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (6)
online/src/main/scala/ai/chronon/online/metrics/OtelMetricsReporter.scala (1)
104-104: Fix duplicate commentRemove the duplicated "// Configure periodic metric reader" text.
- // Configure periodic metric reader// Configure periodic metric reader + // Configure periodic metric readerflink/src/test/scala/ai/chronon/flink/test/deser/SchemaRegistryDeSerSchemaProviderSpec.scala (1)
37-47: Consider deeper schema validationTest confirms schema retrieval but doesn't verify schema structure.
val deserSchema = schemaRegistrySchemaProvider.schema assert(deserSchema != null) +assert(deserSchema.fieldNames.contains("field1")) +assert(deserSchema.fieldNames.contains("field2"))flink/src/main/scala/ai/chronon/flink/deser/SchemaRegistrySerDe.scala (2)
31-32: Consider using boolean default directly.String conversion is unnecessary here when defaulting.
- private val schemaRegistryWireFormat: Boolean = - topicInfo.params.getOrElse(SchemaRegistryWireFormat, "true").toBoolean + private val schemaRegistryWireFormat: Boolean = + topicInfo.params.get(SchemaRegistryWireFormat).map(_.toBoolean).getOrElse(true)
77-86: Optimize byte array handling for large messages.Using
dropcreates a new array which is inefficient for large messages.- val messageBytes = - if (schemaRegistryWireFormat) { - // schema id is set, we skip the first byte and read the schema id based on the wire format: - // https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#messages-wire-format - // unfortunately we need to drop the first 5 bytes (and thus copy the rest of the byte array) as the AvroDataToCatalyst - // interface takes a byte array and the methods to do the Row conversion etc are all private so we can't reach in - message.drop(5) - } else { - message - } + val messageBytes = + if (schemaRegistryWireFormat) { + // schema id is set, we skip the first byte and read the schema id based on the wire format: + // https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#messages-wire-format + // unfortunately we need to drop the first 5 bytes (and thus copy the rest of the byte array) as the AvroDataToCatalyst + // interface takes a byte array and the methods to do the Row conversion etc are all private so we can't reach in + java.util.Arrays.copyOfRange(message, 5, message.length) + } else { + message + }flink/src/main/scala/ai/chronon/flink/deser/DeserializationSchema.scala (2)
77-83: Cache projected schema to avoid redundant evaluator creation.Creating a new evaluator on each call could be inefficient.
+ @transient private lazy val projectedSchemaCache: Array[(String, DataType)] = { + val evaluator = new SparkExpressionEval[Row](sourceEventEncoder, groupBy) + + evaluator.getOutputSchema.fields.map { field => + (field.name, SparkConversions.toChrononType(field.name, field.dataType)) + } + } + override def projectedSchema: Array[(String, DataType)] = { - val evaluator = new SparkExpressionEval[Row](sourceEventEncoder, groupBy) - - evaluator.getOutputSchema.fields.map { field => - (field.name, SparkConversions.toChrononType(field.name, field.dataType)) - } + projectedSchemaCache }
123-128: Consider more specific exception handling.Catching all exceptions might mask specific issues that need different handling.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (31)
cloud_aws/src/main/scala/ai/chronon/integrations/aws/AwsApiImpl.scala(1 hunks)cloud_gcp/BUILD.bazel(1 hunks)cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/GcpApiImpl.scala(2 hunks)cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala(2 hunks)flink/src/main/scala/ai/chronon/flink/FlinkJob.scala(2 hunks)flink/src/main/scala/ai/chronon/flink/KafkaFlinkSource.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/SchemaProvider.scala(0 hunks)flink/src/main/scala/ai/chronon/flink/SchemaRegistrySchemaProvider.scala(0 hunks)flink/src/main/scala/ai/chronon/flink/SparkExpressionEval.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/ChrononDeserializationSchema.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/CustomSchemaSerDe.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/DeserializationSchema.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/FlinkSerDeProvider.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/SchemaRegistrySerDe.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/validation/ValidationFlinkJob.scala(3 hunks)flink/src/main/scala/org/apache/spark/sql/avro/AvroDeserializationSupport.scala(0 hunks)flink/src/test/scala/ai/chronon/flink/test/SchemaRegistrySchemaProviderSpec.scala(0 hunks)flink/src/test/scala/ai/chronon/flink/test/deser/AvroDeSerTestUtils.scala(2 hunks)flink/src/test/scala/ai/chronon/flink/test/deser/SchemaRegistryDeSerSchemaProviderSpec.scala(1 hunks)flink/src/test/scala/ai/chronon/flink/test/deser/SourceIdentityDeSerializationSupportSpec.scala(1 hunks)flink/src/test/scala/ai/chronon/flink/test/deser/SourceProjectionDeSerializationSupportSpec.scala(4 hunks)flink/src/test/scala/org/apache/spark/sql/avro/AvroSourceIdentityDeSerializationSupportSpec.scala(0 hunks)online/src/main/scala/ai/chronon/online/Api.scala(1 hunks)online/src/main/scala/ai/chronon/online/CatalystUtil.scala(1 hunks)online/src/main/scala/ai/chronon/online/metrics/Metrics.scala(1 hunks)online/src/main/scala/ai/chronon/online/metrics/OtelMetricsReporter.scala(1 hunks)online/src/main/scala/ai/chronon/online/serde/AvroSerDe.scala(1 hunks)online/src/main/scala/ai/chronon/online/serde/SerDe.scala(2 hunks)service_commons/src/main/java/ai/chronon/service/ChrononServiceLauncher.java(1 hunks)spark/src/main/scala/ai/chronon/spark/GroupByUpload.scala(1 hunks)spark/src/main/scala/ai/chronon/spark/utils/MockApi.scala(1 hunks)
💤 Files with no reviewable changes (5)
- flink/src/test/scala/ai/chronon/flink/test/SchemaRegistrySchemaProviderSpec.scala
- flink/src/test/scala/org/apache/spark/sql/avro/AvroSourceIdentityDeSerializationSupportSpec.scala
- flink/src/main/scala/ai/chronon/flink/SchemaProvider.scala
- flink/src/main/scala/ai/chronon/flink/SchemaRegistrySchemaProvider.scala
- flink/src/main/scala/org/apache/spark/sql/avro/AvroDeserializationSupport.scala
🧰 Additional context used
🧬 Code Graph Analysis (9)
flink/src/main/scala/ai/chronon/flink/KafkaFlinkSource.scala (2)
flink/src/main/scala/ai/chronon/flink/deser/SchemaRegistrySerDe.scala (1)
flink(34-45)flink/src/main/scala/ai/chronon/flink/deser/ChrononDeserializationSchema.scala (1)
ChrononDeserializationSchema(16-20)
cloud_aws/src/main/scala/ai/chronon/integrations/aws/AwsApiImpl.scala (2)
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/GcpApiImpl.scala (1)
streamDecoder(42-45)spark/src/main/scala/ai/chronon/spark/utils/MockApi.scala (1)
streamDecoder(106-111)
online/src/main/scala/ai/chronon/online/Api.scala (3)
cloud_aws/src/main/scala/ai/chronon/integrations/aws/AwsApiImpl.scala (1)
streamDecoder(45-45)cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/GcpApiImpl.scala (1)
streamDecoder(42-45)spark/src/main/scala/ai/chronon/spark/utils/MockApi.scala (1)
streamDecoder(106-111)
flink/src/main/scala/ai/chronon/flink/SparkExpressionEval.scala (1)
online/src/main/scala/ai/chronon/online/CatalystUtil.scala (4)
performSql(97-99)performSql(131-136)performSql(138-141)performSql(143-147)
service_commons/src/main/java/ai/chronon/service/ChrononServiceLauncher.java (1)
online/src/main/scala/ai/chronon/online/metrics/OtelMetricsReporter.scala (3)
OtelMetricsReporter(19-79)OtelMetricsReporter(81-149)getExporterUrl(93-95)
flink/src/main/scala/ai/chronon/flink/validation/ValidationFlinkJob.scala (4)
flink/src/main/scala/ai/chronon/flink/FlinkSource.scala (1)
FlinkSource(6-16)flink/src/main/scala/ai/chronon/flink/KafkaFlinkSource.scala (1)
KafkaFlinkSource(59-62)flink/src/main/scala/ai/chronon/flink/deser/ChrononDeserializationSchema.scala (2)
DeserializationSchemaBuilder(29-38)buildSourceIdentityDeserSchema(30-32)flink/src/main/scala/ai/chronon/flink/deser/FlinkSerDeProvider.scala (2)
FlinkSerDeProvider(9-26)build(10-25)
online/src/main/scala/ai/chronon/online/metrics/Metrics.scala (1)
online/src/main/scala/ai/chronon/online/metrics/OtelMetricsReporter.scala (4)
OtelMetricsReporter(19-79)OtelMetricsReporter(81-149)buildOtelMetricReader(97-118)buildOpenTelemetryClient(120-148)
cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala (1)
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala (3)
SubmitterConf(16-23)DataprocSubmitter(31-176)DataprocSubmitter(178-312)
online/src/main/scala/ai/chronon/online/metrics/OtelMetricsReporter.scala (1)
api/src/main/java/ai/chronon/api/thrift/Option.java (2)
Option(25-143)Some(93-111)
🔇 Additional comments (47)
service_commons/src/main/java/ai/chronon/service/ChrononServiceLauncher.java (3)
30-30: Robust URL validationChecks if the OTLP exporter URL is defined before initializing metrics, preventing errors when no URL is provided.
32-32: Improved initialization conditionMetrics are now only initialized when both enabled AND URL is defined, reducing log spam.
39-39: Safe URL extractionSafely extracts URL value since we've already checked isDefined().
online/src/main/scala/ai/chronon/online/metrics/Metrics.scala (1)
138-143: Robust OpenTelemetry initializationHandles potential absence of metric reader gracefully, falling back to no-op client.
online/src/main/scala/ai/chronon/online/metrics/OtelMetricsReporter.scala (4)
93-95: Improved URL handlingUses Option pattern for exporter URL, making code more robust against nulls.
97-97: Safer return typeChanged return type to Option[MetricReader] to handle cases where no URL is defined.
101-106: Conditional metric reader creationOnly creates HTTP metric reader when URL exists, using functional style.
111-114: Consistent return typeWraps Prometheus server in Some() to match the Option return type.
cloud_gcp/BUILD.bazel (1)
38-38: Added required Avro dependency.The addition of the Avro library dependency supports the SerDe refactoring in this PR.
flink/src/main/scala/ai/chronon/flink/KafkaFlinkSource.scala (1)
3-3: Added necessary import for ChrononDeserializationSchema.This import supports the constructor parameters in KafkaFlinkSource (line 60) and ProjectedKafkaFlinkSource (line 65).
spark/src/main/scala/ai/chronon/spark/GroupByUpload.scala (1)
271-273: Added empty result validation.Prevents silent failures by validating data is present before extracting metrics.
cloud_aws/src/main/scala/ai/chronon/integrations/aws/AwsApiImpl.scala (1)
45-45: Updated return type from Serde to SerDe.Standardizes the interface naming convention across the codebase.
cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala (2)
79-80: Explicit config improves test reliability.Replacing default constructor with explicit SubmitterConf ensures deterministic test execution independent of external configs.
95-96: Added event delay parameter to test Kafka beacon events.The new
--event-delay-millis=10parameter controls event processing timing, likely needed to test refactored SerDe functionality.flink/src/main/scala/ai/chronon/flink/SparkExpressionEval.scala (1)
95-98: New method enables processing Array[Any] from SerDe.This overload bridges the gap between
Array[Any](from Mutation interfaces) and InternalRow processing, supporting the refactored SerDe architecture.online/src/main/scala/ai/chronon/online/Api.scala (1)
175-175: Return type standardized from Serde to SerDe.Updated return type aligns with PR objective to standardize SerDe interface naming across the codebase.
online/src/main/scala/ai/chronon/online/CatalystUtil.scala (1)
122-122: Made inputArrEncoder public with explicit type.Making this function public allows SparkExpressionEval to use it for Mutation array conversion, supporting the SerDe refactoring.
spark/src/main/scala/ai/chronon/spark/utils/MockApi.scala (1)
106-111: Properly updated to use new SerDe interfaceClean transition from
SerdetoSerDeinterface with proper schema conversion viaAvroConversions.fromChrononSchema().flink/src/test/scala/ai/chronon/flink/test/deser/AvroDeSerTestUtils.scala (2)
1-5: Package and import updates align with new SerDe structurePackage moved from Spark to Flink namespace with proper imports.
22-28: Good implementation of new SerDe interfaceClean implementation of the
SerDeinterface delegating to the underlyingAvroSerDe. This enables in-memory testing of Avro deserialization without schema registry dependencies.online/src/main/scala/ai/chronon/online/serde/SerDe.scala (2)
3-11: Improved SerDe interface with better documentationGood job renaming from
SerdetoSerDeand adding detailed JavaDoc explaining serialization requirements and distributed system considerations.
36-37: Updated documentation referenceDocumentation reference correctly updated to match new class name.
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/GcpApiImpl.scala (2)
10-10: Updated import for new SerDe classesImports properly consolidated and updated to include new SerDe components.
42-43: Updated streamDecoder implementation to use new SerDe interfaceReturn type properly changed from
SerdetoSerDewith implementation updated to use schema conversion before instantiatingAvroSerDe.flink/src/main/scala/ai/chronon/flink/validation/ValidationFlinkJob.scala (3)
5-5: Updated imports to match new SerDe architecture.Clean import refactoring to use the new SerDe interfaces.
Also applies to: 21-21
142-142: Refactored to use FlinkSerDeProvider.Improved schema provider creation by delegating to centralized factory.
144-145: Using builder pattern for deserialization schema.Properly uses the DeserializationSchemaBuilder to create a SourceIdentityDeserSchema.
flink/src/main/scala/ai/chronon/flink/deser/FlinkSerDeProvider.scala (2)
1-7: New SerDe provider package structure.Clean imports and package organization.
8-26: Well-structured SerDe factory implementation.The implementation:
- Enforces mutual exclusivity of configuration options
- Provides clear error messages
- Delegates to appropriate SerDe implementations
flink/src/main/scala/ai/chronon/flink/FlinkJob.scala (3)
11-11: Updated imports to match new SerDe architecture.Clean import refactoring to use the new SerDe interfaces.
315-315: Refactored to use FlinkSerDeProvider.Improved schema provider creation by delegating to centralized factory.
317-318: Using builder pattern for deserialization schema.Properly uses the DeserializationSchemaBuilder to create a SourceProjectionDeserSchema.
flink/src/main/scala/ai/chronon/flink/deser/CustomSchemaSerDe.scala (2)
1-7: New custom SerDe provider implementation.Clean package structure with helpful example in comment.
8-21: Well-implemented dynamic SerDe loading.The implementation:
- Properly handles configuration validation
- Uses reflection correctly to load custom provider classes
- Has clear error messages for misconfiguration
flink/src/test/scala/ai/chronon/flink/test/deser/SourceIdentityDeSerializationSupportSpec.scala (1)
14-67: Well-structured test cases for identity deserializationTests verify both successful deserialization and error handling when data is corrupted.
online/src/main/scala/ai/chronon/online/serde/AvroSerDe.scala (1)
24-35: Implementation aligns with refactor goalsSuccessfully handles Mutation types through SerDe classes.
flink/src/test/scala/ai/chronon/flink/test/deser/SchemaRegistryDeSerSchemaProviderSpec.scala (3)
28-35: Test validates failure case properlyGood test for schema subject not found scenario.
49-60: Validates custom subject usageTests confirm injection of custom subjects works correctly.
62-73: Good error case coverageAppropriate validation for unsupported schema types.
flink/src/test/scala/ai/chronon/flink/test/deser/SourceProjectionDeSerializationSupportSpec.scala (3)
26-28: Successfully refactored to use modular approachUpdated implementation follows new SerDe pattern.
65-67: Consistent implementation of updated patternMatches updated approach across all test cases.
88-90: Correct implementation of error handling testFinal test case properly validates behavior with corrupt data.
flink/src/main/scala/ai/chronon/flink/deser/ChrononDeserializationSchema.scala (3)
16-20: Well-designed abstraction and clean interface.The abstract class defines a clear contract for all deserialization schemas with appropriate abstract methods.
25-27: Clean trait definition for projection support.Good use of a trait to mark schemas supporting projection pushdown.
29-38: Effective builder pattern implementation.Builder methods create appropriate schema instances, promoting clean factory pattern usage.
flink/src/main/scala/ai/chronon/flink/deser/DeserializationSchema.scala (2)
61-64: Good enforcement of collector-based deserialization.Forces correct method usage through explicit exception.
132-144: Efficient implementation of row handling.Good use of while loop for performance in array conversion.
Comment on lines
+17
to
+22
| private def byteArrayToAvro(avro: Array[Byte], schema: Schema): GenericRecord = { | ||
| val reader = new SpecificDatumReader[GenericRecord](schema) | ||
| val input: InputStream = new ByteArrayInputStream(avro) | ||
| val decoder: BinaryDecoder = DecoderFactory.get().binaryDecoder(input, null) | ||
| reader.read(null, decoder) | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add error handling for corrupt data
Method should handle exceptions when byte array doesn't match schema.
private def byteArrayToAvro(avro: Array[Byte], schema: Schema): GenericRecord = {
- val reader = new SpecificDatumReader[GenericRecord](schema)
- val input: InputStream = new ByteArrayInputStream(avro)
- val decoder: BinaryDecoder = DecoderFactory.get().binaryDecoder(input, null)
- reader.read(null, decoder)
+ try {
+ val reader = new SpecificDatumReader[GenericRecord](schema)
+ val input: InputStream = new ByteArrayInputStream(avro)
+ val decoder: BinaryDecoder = DecoderFactory.get().binaryDecoder(input, null)
+ reader.read(null, decoder)
+ } catch {
+ case e: Exception =>
+ throw new IllegalArgumentException(s"Failed to deserialize Avro data: ${e.getMessage}", e)
+ }
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| private def byteArrayToAvro(avro: Array[Byte], schema: Schema): GenericRecord = { | |
| val reader = new SpecificDatumReader[GenericRecord](schema) | |
| val input: InputStream = new ByteArrayInputStream(avro) | |
| val decoder: BinaryDecoder = DecoderFactory.get().binaryDecoder(input, null) | |
| reader.read(null, decoder) | |
| } | |
| private def byteArrayToAvro(avro: Array[Byte], schema: Schema): GenericRecord = { | |
| try { | |
| val reader = new SpecificDatumReader[GenericRecord](schema) | |
| val input: InputStream = new ByteArrayInputStream(avro) | |
| val decoder: BinaryDecoder = DecoderFactory.get().binaryDecoder(input, null) | |
| reader.read(null, decoder) | |
| } catch { | |
| case e: Exception => | |
| throw new IllegalArgumentException(s"Failed to deserialize Avro data: ${e.getMessage}", e) | |
| } | |
| } |
piyush-zlai
force-pushed
the
piyush/flink_mutations
branch
from
5a23f82 to
28509fb
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
flink/src/test/scala/ai/chronon/flink/test/deser/SourceIdentityDeSerializationSupportSpec.scala (2)
44-47: Consider expanding field verification.Only checks two fields (id and username). Consider verifying more fields for complete coverage.
59-60: Simple corruption approach.Consider testing different corruption scenarios (middle of data, end of data) to ensure resilience in various failure modes.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (30)
cloud_aws/src/main/scala/ai/chronon/integrations/aws/AwsApiImpl.scala(1 hunks)cloud_gcp/BUILD.bazel(1 hunks)cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/GcpApiImpl.scala(2 hunks)cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala(2 hunks)flink/src/main/scala/ai/chronon/flink/FlinkJob.scala(2 hunks)flink/src/main/scala/ai/chronon/flink/KafkaFlinkSource.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/SchemaProvider.scala(0 hunks)flink/src/main/scala/ai/chronon/flink/SchemaRegistrySchemaProvider.scala(0 hunks)flink/src/main/scala/ai/chronon/flink/SparkExpressionEval.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/ChrononDeserializationSchema.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/CustomSchemaSerDe.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/DeserializationSchema.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/FlinkSerDeProvider.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/deser/SchemaRegistrySerDe.scala(1 hunks)flink/src/main/scala/ai/chronon/flink/validation/ValidationFlinkJob.scala(3 hunks)flink/src/main/scala/org/apache/spark/sql/avro/AvroDeserializationSupport.scala(0 hunks)flink/src/test/scala/ai/chronon/flink/test/SchemaRegistrySchemaProviderSpec.scala(0 hunks)flink/src/test/scala/ai/chronon/flink/test/deser/AvroDeSerTestUtils.scala(2 hunks)flink/src/test/scala/ai/chronon/flink/test/deser/SchemaRegistryDeSerSchemaProviderSpec.scala(1 hunks)flink/src/test/scala/ai/chronon/flink/test/deser/SourceIdentityDeSerializationSupportSpec.scala(1 hunks)flink/src/test/scala/ai/chronon/flink/test/deser/SourceProjectionDeSerializationSupportSpec.scala(4 hunks)flink/src/test/scala/org/apache/spark/sql/avro/AvroSourceIdentityDeSerializationSupportSpec.scala(0 hunks)online/src/main/scala/ai/chronon/online/Api.scala(1 hunks)online/src/main/scala/ai/chronon/online/CatalystUtil.scala(1 hunks)online/src/main/scala/ai/chronon/online/metrics/Metrics.scala(1 hunks)online/src/main/scala/ai/chronon/online/metrics/OtelMetricsReporter.scala(1 hunks)online/src/main/scala/ai/chronon/online/serde/AvroSerDe.scala(1 hunks)online/src/main/scala/ai/chronon/online/serde/SerDe.scala(2 hunks)service_commons/src/main/java/ai/chronon/service/ChrononServiceLauncher.java(1 hunks)spark/src/main/scala/ai/chronon/spark/utils/MockApi.scala(1 hunks)
💤 Files with no reviewable changes (5)
- flink/src/test/scala/ai/chronon/flink/test/SchemaRegistrySchemaProviderSpec.scala
- flink/src/test/scala/org/apache/spark/sql/avro/AvroSourceIdentityDeSerializationSupportSpec.scala
- flink/src/main/scala/ai/chronon/flink/SchemaProvider.scala
- flink/src/main/scala/ai/chronon/flink/SchemaRegistrySchemaProvider.scala
- flink/src/main/scala/org/apache/spark/sql/avro/AvroDeserializationSupport.scala
🚧 Files skipped from review as they are similar to previous changes (24)
- cloud_gcp/BUILD.bazel
- flink/src/main/scala/ai/chronon/flink/KafkaFlinkSource.scala
- flink/src/main/scala/ai/chronon/flink/SparkExpressionEval.scala
- online/src/main/scala/ai/chronon/online/Api.scala
- spark/src/main/scala/ai/chronon/spark/utils/MockApi.scala
- flink/src/main/scala/ai/chronon/flink/FlinkJob.scala
- online/src/main/scala/ai/chronon/online/metrics/Metrics.scala
- online/src/main/scala/ai/chronon/online/serde/SerDe.scala
- cloud_aws/src/main/scala/ai/chronon/integrations/aws/AwsApiImpl.scala
- flink/src/main/scala/ai/chronon/flink/validation/ValidationFlinkJob.scala
- flink/src/test/scala/ai/chronon/flink/test/deser/AvroDeSerTestUtils.scala
- cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/GcpApiImpl.scala
- online/src/main/scala/ai/chronon/online/metrics/OtelMetricsReporter.scala
- flink/src/main/scala/ai/chronon/flink/deser/CustomSchemaSerDe.scala
- flink/src/test/scala/ai/chronon/flink/test/deser/SchemaRegistryDeSerSchemaProviderSpec.scala
- online/src/main/scala/ai/chronon/online/CatalystUtil.scala
- service_commons/src/main/java/ai/chronon/service/ChrononServiceLauncher.java
- online/src/main/scala/ai/chronon/online/serde/AvroSerDe.scala
- flink/src/main/scala/ai/chronon/flink/deser/FlinkSerDeProvider.scala
- flink/src/test/scala/ai/chronon/flink/test/deser/SourceProjectionDeSerializationSupportSpec.scala
- flink/src/main/scala/ai/chronon/flink/deser/SchemaRegistrySerDe.scala
- cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala
- flink/src/main/scala/ai/chronon/flink/deser/ChrononDeserializationSchema.scala
- flink/src/main/scala/ai/chronon/flink/deser/DeserializationSchema.scala
⏰ Context from checks skipped due to timeout of 90000ms (28)
- GitHub Check: service_commons_tests
- GitHub Check: streaming_tests
- GitHub Check: streaming_tests
- GitHub Check: service_tests
- GitHub Check: service_tests
- GitHub Check: join_tests
- GitHub Check: cloud_gcp_tests
- GitHub Check: groupby_tests
- GitHub Check: analyzer_tests
- GitHub Check: cloud_gcp_tests
- GitHub Check: groupby_tests
- GitHub Check: cloud_aws_tests
- GitHub Check: cloud_aws_tests
- GitHub Check: join_tests
- GitHub Check: fetcher_tests
- GitHub Check: api_tests
- GitHub Check: fetcher_tests
- GitHub Check: analyzer_tests
- GitHub Check: api_tests
- GitHub Check: online_tests
- GitHub Check: batch_tests
- GitHub Check: flink_tests
- GitHub Check: flink_tests
- GitHub Check: spark_tests
- GitHub Check: batch_tests
- GitHub Check: spark_tests
- GitHub Check: online_tests
- GitHub Check: enforce_triggered_workflows
🔇 Additional comments (4)
flink/src/test/scala/ai/chronon/flink/test/deser/SourceIdentityDeSerializationSupportSpec.scala (4)
1-13: Appropriate imports for test functionality.Clean import organization with necessary dependencies.
17-47: Good test for happy path deserialization.Test verifies schema alignment and basic data integrity after deserialization.
49-66: Good error resilience test.Properly tests that corrupted Avro data is handled gracefully.
14-67: Test class aligns with PR objectives.Appropriately validates the new deserialization infrastructure using the SerDe interfaces as intended.
This was referenced May 12, 2025
Merged
piyush-zlai
added a commit
that referenced
this pull request
May 12, 2025
## Summary Pulling out from PR - #751 as we're waiting on an r there and it shows up as noise in various places so lets just fix. ## Checklist - [ ] Added Unit Tests - [ ] Covered by existing CI - [ ] Integration tested - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **Bug Fixes** - Improved handling of metrics exporter URL configuration to prevent errors when the URL is not defined. - Ensured metrics are only initialized when both metrics are enabled and an exporter URL is present. - **Refactor** - Enhanced internal logic for safer initialization of metrics reporting, reducing the risk of misconfiguration. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
tchow-zlai
approved these changes
May 13, 2025
piyush-zlai
added a commit
that referenced
this pull request
May 13, 2025
## Summary Builds on top of PR: #751. This PR adds a streaming GroupBy that can be run as a canary to sanity check and test things out while making Flink changes. I used this to sanity check the creation & use of a Mock schema serde that some users have been asking for. Can be submitted via: ``` $ CHRONON_ROOT=`pwd`/api/python/test/canary $ zipline compile --chronon-root=$CHRONON_ROOT $ zipline run --repo=$CHRONON_ROOT --version $VERSION --mode streaming --conf compiled/group_bys/gcp/item_event_canary.actions_v1 --kafka-bootstrap=bootstrap.zipline-kafka-cluster.us-central1.managedkafka.canary-443022.cloud.goog:9092 --groupby-name gcp.item_event_canary.actions_v1 --validate ``` (Needs the Flink event driver to be running - triggered via DataProcSubmitterTest) ## Checklist - [ ] Added Unit Tests - [ ] Covered by existing CI - [X] Integration tested - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit ## Summary by CodeRabbit - **New Features** - Introduced a new group-by aggregation for item event actions, supporting real-time analytics by listing ID with data sourced from GCP Kafka and BigQuery. - Added a mock schema provider for testing item event ingestion. - **Bug Fixes** - Updated test configurations to use new event schemas, topics, and data paths for improved accuracy in Flink Kafka ingest job tests. - **Refactor** - Renamed and restructured the event driver to focus on item events, with a streamlined schema and updated job naming. - **Chores** - Added new environment variable for Flink state storage configuration. - Updated build configuration to reference the renamed event driver. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
chewy-zlai
pushed a commit
that referenced
this pull request
May 15, 2025
## Summary Pulling out from PR - #751 as we're waiting on an r there and it shows up as noise in various places so lets just fix. ## Checklist - [ ] Added Unit Tests - [ ] Covered by existing CI - [ ] Integration tested - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **Bug Fixes** - Improved handling of metrics exporter URL configuration to prevent errors when the URL is not defined. - Ensured metrics are only initialized when both metrics are enabled and an exporter URL is present. - **Refactor** - Enhanced internal logic for safer initialization of metrics reporting, reducing the risk of misconfiguration. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
chewy-zlai
pushed a commit
that referenced
this pull request
May 15, 2025
…rfaces (#751) ## Summary Refactor some of the schema provider shaped code to - * Use the existing SerDe class interfaces we have * Work with Mutation types via the SerDe classes * Primary shuffling is around pulling the Avro deser out of the existing BaseAvroDeserializationSchema and delegating that to the SerDe to get a Mutation back as well as shifting things a bit to call CatalystUtil with the Mutation Array[Any] types. * Provide rails for users to provide a custom schema provider. I used this to test a version of the beacon app out in canary - I'll put up a separate PR for the test job in a follow up. * Other misc piled up fixes - Check that GBUs don't compute empty results; fix our Otel metrics code to be turned off by default to reduce log spam. ## Checklist - [X] Added Unit Tests - [X] Covered by existing CI - [X] Integration tested -- Tested via canary on our env / cust env and confirmed we pass the validation piece as well as see the jobs come up and write out data to BT. - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **New Features** - Added Avro serialization and deserialization support for online data processing. - Introduced flexible schema registry and custom schema provider selection for Flink streaming sources. - **Refactor** - Unified and renamed the serialization/deserialization interface to `SerDe` across modules. - Centralized and simplified schema provider and deserialization logic for Flink jobs. - Improved visibility and type safety for internal utilities. - **Bug Fixes** - Enhanced error handling and robustness in metrics initialization and deserialization workflows. - **Tests** - Added and updated tests for Avro deserialization and schema registry integration. - Removed outdated or redundant test suites. - **Chores** - Updated external dependencies to include Avro support. - Cleaned up unused files and legacy code. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
chewy-zlai
pushed a commit
that referenced
this pull request
May 15, 2025
## Summary Builds on top of PR: #751. This PR adds a streaming GroupBy that can be run as a canary to sanity check and test things out while making Flink changes. I used this to sanity check the creation & use of a Mock schema serde that some users have been asking for. Can be submitted via: ``` $ CHRONON_ROOT=`pwd`/api/python/test/canary $ zipline compile --chronon-root=$CHRONON_ROOT $ zipline run --repo=$CHRONON_ROOT --version $VERSION --mode streaming --conf compiled/group_bys/gcp/item_event_canary.actions_v1 --kafka-bootstrap=bootstrap.zipline-kafka-cluster.us-central1.managedkafka.canary-443022.cloud.goog:9092 --groupby-name gcp.item_event_canary.actions_v1 --validate ``` (Needs the Flink event driver to be running - triggered via DataProcSubmitterTest) ## Checklist - [ ] Added Unit Tests - [ ] Covered by existing CI - [X] Integration tested - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit ## Summary by CodeRabbit - **New Features** - Introduced a new group-by aggregation for item event actions, supporting real-time analytics by listing ID with data sourced from GCP Kafka and BigQuery. - Added a mock schema provider for testing item event ingestion. - **Bug Fixes** - Updated test configurations to use new event schemas, topics, and data paths for improved accuracy in Flink Kafka ingest job tests. - **Refactor** - Renamed and restructured the event driver to focus on item events, with a streamlined schema and updated job naming. - **Chores** - Added new environment variable for Flink state storage configuration. - Updated build configuration to reference the renamed event driver. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
chewy-zlai
pushed a commit
that referenced
this pull request
May 15, 2025
## Summary Pulling out from PR - #751 as we're waiting on an r there and it shows up as noise in various places so lets just fix. ## Checklist - [ ] Added Unit Tests - [ ] Covered by existing CI - [ ] Integration tested - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **Bug Fixes** - Improved handling of metrics exporter URL configuration to prevent errors when the URL is not defined. - Ensured metrics are only initialized when both metrics are enabled and an exporter URL is present. - **Refactor** - Enhanced internal logic for safer initialization of metrics reporting, reducing the risk of misconfiguration. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
chewy-zlai
pushed a commit
that referenced
this pull request
May 15, 2025
…rfaces (#751) ## Summary Refactor some of the schema provider shaped code to - * Use the existing SerDe class interfaces we have * Work with Mutation types via the SerDe classes * Primary shuffling is around pulling the Avro deser out of the existing BaseAvroDeserializationSchema and delegating that to the SerDe to get a Mutation back as well as shifting things a bit to call CatalystUtil with the Mutation Array[Any] types. * Provide rails for users to provide a custom schema provider. I used this to test a version of the beacon app out in canary - I'll put up a separate PR for the test job in a follow up. * Other misc piled up fixes - Check that GBUs don't compute empty results; fix our Otel metrics code to be turned off by default to reduce log spam. ## Checklist - [X] Added Unit Tests - [X] Covered by existing CI - [X] Integration tested -- Tested via canary on our env / cust env and confirmed we pass the validation piece as well as see the jobs come up and write out data to BT. - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **New Features** - Added Avro serialization and deserialization support for online data processing. - Introduced flexible schema registry and custom schema provider selection for Flink streaming sources. - **Refactor** - Unified and renamed the serialization/deserialization interface to `SerDe` across modules. - Centralized and simplified schema provider and deserialization logic for Flink jobs. - Improved visibility and type safety for internal utilities. - **Bug Fixes** - Enhanced error handling and robustness in metrics initialization and deserialization workflows. - **Tests** - Added and updated tests for Avro deserialization and schema registry integration. - Removed outdated or redundant test suites. - **Chores** - Updated external dependencies to include Avro support. - Cleaned up unused files and legacy code. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
chewy-zlai
pushed a commit
that referenced
this pull request
May 15, 2025
## Summary Builds on top of PR: #751. This PR adds a streaming GroupBy that can be run as a canary to sanity check and test things out while making Flink changes. I used this to sanity check the creation & use of a Mock schema serde that some users have been asking for. Can be submitted via: ``` $ CHRONON_ROOT=`pwd`/api/python/test/canary $ zipline compile --chronon-root=$CHRONON_ROOT $ zipline run --repo=$CHRONON_ROOT --version $VERSION --mode streaming --conf compiled/group_bys/gcp/item_event_canary.actions_v1 --kafka-bootstrap=bootstrap.zipline-kafka-cluster.us-central1.managedkafka.canary-443022.cloud.goog:9092 --groupby-name gcp.item_event_canary.actions_v1 --validate ``` (Needs the Flink event driver to be running - triggered via DataProcSubmitterTest) ## Checklist - [ ] Added Unit Tests - [ ] Covered by existing CI - [X] Integration tested - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit ## Summary by CodeRabbit - **New Features** - Introduced a new group-by aggregation for item event actions, supporting real-time analytics by listing ID with data sourced from GCP Kafka and BigQuery. - Added a mock schema provider for testing item event ingestion. - **Bug Fixes** - Updated test configurations to use new event schemas, topics, and data paths for improved accuracy in Flink Kafka ingest job tests. - **Refactor** - Renamed and restructured the event driver to focus on item events, with a streamlined schema and updated job naming. - **Chores** - Added new environment variable for Flink state storage configuration. - Updated build configuration to reference the renamed event driver. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
chewy-zlai
pushed a commit
that referenced
this pull request
May 16, 2025
## Summary Pulling out from PR - #751 as we're waiting on an r there and it shows up as noise in various places so lets just fix. ## Cheour clientslist - [ ] Added Unit Tests - [ ] Covered by existing CI - [ ] Integration tested - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **Bug Fixes** - Improved handling of metrics exporter URL configuration to prevent errors when the URL is not defined. - Ensured metrics are only initialized when both metrics are enabled and an exporter URL is present. - **Refactor** - Enhanced internal logic for safer initialization of metrics reporting, reducing the risk of misconfiguration. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
chewy-zlai
pushed a commit
that referenced
this pull request
May 16, 2025
…rfaces (#751) ## Summary Refactor some of the schema provider shaped code to - * Use the existing SerDe class interfaces we have * Work with Mutation types via the SerDe classes * Primary shuffling is around pulling the Avro deser out of the existing BaseAvroDeserializationSchema and delegating that to the SerDe to get a Mutation baour clients as well as shifting things a bit to call CatalystUtil with the Mutation Array[Any] types. * Provide rails for users to provide a custom schema provider. I used this to test a version of the beacon app out in canary - I'll put up a separate PR for the test job in a follow up. * Other misc piled up fixes - Cheour clients that GBUs don't compute empty results; fix our Otel metrics code to be turned off by default to reduce log spam. ## Cheour clientslist - [X] Added Unit Tests - [X] Covered by existing CI - [X] Integration tested -- Tested via canary on our env / cust env and confirmed we pass the validation piece as well as see the jobs come up and write out data to BT. - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **New Features** - Added Avro serialization and deserialization support for online data processing. - Introduced flexible schema registry and custom schema provider selection for Flink streaming sources. - **Refactor** - Unified and renamed the serialization/deserialization interface to `SerDe` across modules. - Centralized and simplified schema provider and deserialization logic for Flink jobs. - Improved visibility and type safety for internal utilities. - **Bug Fixes** - Enhanced error handling and robustness in metrics initialization and deserialization workflows. - **Tests** - Added and updated tests for Avro deserialization and schema registry integration. - Removed outdated or redundant test suites. - **Chores** - Updated external dependencies to include Avro support. - Cleaned up unused files and legacy code. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
chewy-zlai
pushed a commit
that referenced
this pull request
May 16, 2025
## Summary Builds on top of PR: #751. This PR adds a streaming GroupBy that can be run as a canary to sanity cheour clients and test things out while making Flink changes. I used this to sanity cheour clients the creation & use of a Moour clients schema serde that some users have been asking for. Can be submitted via: ``` $ CHRONON_ROOT=`pwd`/api/python/test/canary $ zipline compile --chronon-root=$CHRONON_ROOT $ zipline run --repo=$CHRONON_ROOT --version $VERSION --mode streaming --conf compiled/group_bys/gcp/item_event_canary.actions_v1 --kafka-bootstrap=bootstrap.zipline-kafka-cluster.us-central1.managedkafka.canary-443022.cloud.goog:9092 --groupby-name gcp.item_event_canary.actions_v1 --validate ``` (Needs the Flink event driver to be running - triggered via DataProcSubmitterTest) ## Cheour clientslist - [ ] Added Unit Tests - [ ] Covered by existing CI - [X] Integration tested - [ ] Documentation update <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit ## Summary by CodeRabbit - **New Features** - Introduced a new group-by aggregation for item event actions, supporting real-time analytics by listing ID with data sourced from GCP Kafka and BigQuery. - Added a moour clients schema provider for testing item event ingestion. - **Bug Fixes** - Updated test configurations to use new event schemas, topics, and data paths for improved accuracy in Flink Kafka ingest job tests. - **Refactor** - Renamed and restructured the event driver to focus on item events, with a streamlined schema and updated job naming. - **Chores** - Added new environment variable for Flink state storage configuration. - Updated build configuration to reference the renamed event driver. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
Refactor some of the schema provider shaped code to -
Checklist
-- Tested via canary on our env / cust env and confirmed we pass the validation piece as well as see the jobs come up and write out data to BT.
Summary by CodeRabbit
SerDeacross modules.