kumarpritam863 commented on code in PR #18739:
URL: https://github.com/apache/iceberg/pull/18739#discussion_r4520128207
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicTableUpdateOperator.java:
##########
@@ -271,6 +285,126 @@ void testDynamicTableUpdateOperatorDropUnusedColumns()
throws Exception {
assertThat(input).isEqualTo(output);
}
+ @Test
+ void testDynamicTableUpdateOperatorErrorStream() throws Exception {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier table = TableIdentifier.of(TABLE);
+
+ AtomicInteger errorCount = new AtomicInteger();
+ AtomicReference<Tuple2<DynamicRecordInternal, Exception>> errorRecord =
new AtomicReference<>();
+
+ DynamicTableUpdateOperator operator =
+ new DynamicTableUpdateOperator(
+ CATALOG_EXTENSION.catalogLoader(),
+ TableCreator.DEFAULT,
+ flinkDynamicSinkConfiguration(CASE_SENSITIVE, PRESERVE_COLUMNS),
+ error -> true);
+ operator.open(null);
+
+ catalog.createTable(table, SCHEMA1);
+
+ // Incompatible evolution (int -> boolean) fails the update; the filter
whitelists it, so the
+ // record is routed to the side output instead of failing the job.
+ Schema incompatibleSchema =
+ new Schema(Types.NestedField.required(1, "id",
Types.BooleanType.get()));
+ DynamicRecordInternal input =
+ new DynamicRecordInternal(
+ TABLE,
+ "branch",
+ incompatibleSchema,
+ GenericRowData.of(true),
+ PartitionSpec.unpartitioned(),
+ 42,
+ false,
+ Collections.emptySet());
+
+ TestCollector collector = new TestCollector();
+ ProcessFunction<DynamicRecordInternal, DynamicRecordInternal>.Context
mockContext =
+ Mockito.mock(ProcessFunction.Context.class);
Review Comment:
Switched the error stream tests to
ProcessFunctionTestHarnesses.forProcessFunction and getSideOutput, and Mockito
is gone from the class. Added failureAfterTheTableWasUpdatedIsDiverted. The
older tests that don't use the error stream still call processElement directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
