Copilot commented on code in PR #2536:
URL: https://github.com/apache/phoenix/pull/2536#discussion_r3789335902
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java:
##########
@@ -149,9 +158,23 @@ protected void setup(Context context) throws IOException,
InterruptedException {
}
ignoreInvalidRows = conf.getBoolean(IGNORE_INVALID_ROW_CONFKEY, true);
- upsertListener = new MapperUpsertListener<RECORD>(context,
ignoreInvalidRows);
+ upsertListener = new MapperUpsertListener<RECORD>(context,
ignoreInvalidRows,
+ this::writeBadRecord);
upsertExecutor = buildUpsertExecutor(conf);
preUpdateProcessor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
+
+ String badRecordsPath = conf.get(BAD_RECORDS_PATH_CONFKEY);
+ if (badRecordsPath != null) {
+ Path outputDir = new Path(badRecordsPath);
+ FileSystem fs = outputDir.getFileSystem(conf);
+ if (!fs.exists(outputDir)) {
+ fs.mkdirs(outputDir);
+ }
+ String taskAttemptId = context.getTaskAttemptID().toString();
+ Path badRecordFile = new Path(outputDir, taskAttemptId + ".bad");
+ badRecordsWriter = new PrintWriter(
+ new OutputStreamWriter(fs.create(badRecordFile, false),
StandardCharsets.UTF_8));
Review Comment:
This writes attempt output directly into the final directory, bypassing
MapReduce's output commit protocol. Failed or speculative attempts leave their
`.bad` files behind, so retries can make the audit trail contain duplicate or
non-final rejected rows. Write into attempt work paths and publish only
committed task output via an `OutputCommitter`.
##########
phoenix-core/src/test/java/org/apache/phoenix/mapreduce/BadRecordsWriterTest.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.phoenix.mapreduce;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
Review Comment:
This unused import violates the repository's `UnusedImports` check
(`src/main/config/checkstyle/checker.xml:129`) and will fail style validation.
Remove it.
This issue also appears on line 25 of the same file.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java:
##########
@@ -406,6 +456,9 @@ public void upsertDone(long upsertCount) {
public void errorOnRecord(T record, Throwable throwable) {
LOGGER.error("Error on record " + record, throwable);
context.getCounter(COUNTER_GROUP_NAME, "Errors on
records").increment(1L);
+ if (badRecordWriter != null) {
+ badRecordWriter.write(String.valueOf(record), throwable.getMessage());
Review Comment:
For upsert failures, `record` is already a parsed `CSVRecord` or `Map`, so
`String.valueOf(record)` writes a debug/Java-map representation rather than the
original CSV/JSON input. These files therefore cannot reliably be corrected and
re-imported as promised. Preserve the raw `Text` value in `map` and pass that
raw line to the bad-record writer when the executor reports an error.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java:
##########
@@ -337,9 +361,24 @@ private void writeAggregatedRow(Context context, String
tableName, List<Cell> lk
}
}
+ /**
+ * Write a rejected record to the bad records file if configured.
+ */
+ private void writeBadRecord(String record, String errorMessage) {
+ if (badRecordsWriter != null) {
+ String sanitizedError = errorMessage != null
+ ? errorMessage.replace('\n', ' ').replace('\t', ' ')
+ : "unknown";
+ badRecordsWriter.println(sanitizedError + "\t" + record);
+ }
+ }
+
@Override
protected void cleanup(Context context) throws IOException,
InterruptedException {
try {
+ if (badRecordsWriter != null) {
+ badRecordsWriter.close();
+ }
Review Comment:
`PrintWriter` suppresses write, flush, and close `IOException`s, so HDFS
failures can silently produce a missing or truncated audit file while the task
succeeds. Check its error state after closing and fail cleanup when persistence
failed.
--
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]
