kumarpritam863 commented on code in PR #18739:
URL: https://github.com/apache/iceberg/pull/18739#discussion_r4520121740
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -2143,6 +2145,80 @@ void
testWriteOldShapeRecordAfterColumnReAddedWithWiderType() throws Exception {
assertThat(records.get(0).getField("c")).isEqualTo(5L);
}
+ @Test
+ void testSideOutputForErroneousRecordsImmediateUpdatePath() {
+ runTestSideOutputForErroneousRecords(true);
+ }
+
+ @Test
+ void testSideOutputForErroneousRecordsNonImmediateUpdatePath() {
+ runTestSideOutputForErroneousRecords(false);
+ }
+
+ private void runTestSideOutputForErroneousRecords(boolean immediateUpdate) {
+ // Converting int -> boolean will fail
+ Schema incompatibleSchemaChange =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.BooleanType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+
+ List<DynamicIcebergDataImpl> rows =
+ Lists.newArrayList(
+ new DynamicIcebergDataImpl(
+ SimpleDataUtil.SCHEMA,
+ "t1",
+ SnapshotRef.MAIN_BRANCH,
+ PartitionSpec.unpartitioned()),
+ new DynamicIcebergDataImpl(
+ // Converting int -> boolean will fail
+ incompatibleSchemaChange,
+ "t1",
+ SnapshotRef.MAIN_BRANCH,
+ PartitionSpec.unpartitioned(),
+ true,
+ Collections.singleton("id"),
+ false));
+
+ DataStream<DynamicIcebergDataImpl> dataStream =
+ env.fromData(rows, TypeInformation.of(new TypeHint<>() {}));
+ env.setParallelism(1);
+
+ AtomicInteger errorCount = new AtomicInteger();
+ AtomicReference<Tuple2<DynamicRecord, Exception>> rejectedRecord = new
AtomicReference<>();
+
+ DynamicIcebergSink.forInput(dataStream)
+ .generator(new Generator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .errorStreamConsumer(
+ errorStream -> {
+ // This code runs at the end of the append() call to execute the
Flink job and
+ // retrieve the error stream.
+ // The try/catch is only required due to the checked exception
of executeAndCollect().
+ try {
+ errorStream
+ .executeAndCollect()
+ .forEachRemaining(
+ record -> {
+ errorCount.incrementAndGet();
+ rejectedRecord.set(record);
+ });
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .errorStreamFilter(error -> true)
+ .immediateTableUpdate(immediateUpdate)
+ .append();
+
+ assertThat(errorCount.get()).isEqualTo(1);
Review Comment:
Done. The test now verifies that the first record is written to t1 and that
the exception is a DynamicRecordException. It passes for both update modes.
--
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]
