lokeshj1703 commented on code in PR #7982: URL: https://github.com/apache/hudi/pull/7982#discussion_r1118696059
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/BaseErrorTableWriter.java: ########## @@ -0,0 +1,70 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.utilities.deltastreamer; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.model.HoodieAvroRecord; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.VisibleForTesting; + +import org.apache.hadoop.fs.FileSystem; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.SparkSession; + +/** + * The class which handles error events when delta streamer syncs data from source. All the + * records which Delta streamer is not able to process are triggered as error events to + * BaseErrorTableWriter. The implementation of BaseErrorTableWriter processes + * these error events through addErrorEvents API and commits them to the error table when + * upsertAndCommit API is called. + * + * The writer can use the configs defined in HoodieErrorTableConfig to manage the error table. + */ +public abstract class BaseErrorTableWriter<T extends ErrorEvent> { + + // The column name passed to Spark for option `columnNameOfCorruptRecord`. The record + // is set to this column in case of an error + public static String ERROR_TABLE_CURRUPT_RECORD_COL_NAME = "_corrupt_record"; + + public BaseErrorTableWriter(HoodieDeltaStreamer.Config cfg, SparkSession sparkSession, + TypedProperties props, JavaSparkContext jssc, FileSystem fs) { + } + + /** + * Processes input error events. These error events would be committed later through upsertAndCommit + * API call. + * + * @param errorEvent Input error event RDD + */ + public abstract void addErrorEvents(JavaRDD<T> errorEvent); + + /** + * Fetches the error events RDD processed by the writer so far. This is a test API. + */ + @VisibleForTesting + public abstract Option<JavaRDD<HoodieAvroRecord>> getErrorEvents(String baseTableInstantTime, Option<String> commitedInstantTime); + + /** + * This API is called to commit the error events (failed Hoodie Records) processed by the writer so far. + * These records are committed to a error table. + */ + public abstract boolean upsertAndCommit(String baseTableInstantTime, Option<String> commitedInstantTime); Review Comment: Created https://issues.apache.org/jira/browse/HUDI-5858 for it. -- 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]
