@Test public void testAbort() throws IOException { HoodieWriteConfig cfg = getConfigBuilder(basePath).build(); String instantTime0 = "00" + 0; HoodieDataSourceInternalWriter dataSourceInternalWriter = new HoodieDataSourceInternalWriter(instantTime0, cfg, STRUCT_TYPE, sqlContext.sparkSession(), hadoopConf); DataWriter<InternalRow> writer = dataSourceInternalWriter.createWriterFactory().createDataWriter(0, RANDOM.nextLong(), RANDOM.nextLong()); List<String> partitionPaths = Arrays.asList(HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS); List<String> partitionPathsAbs = new ArrayList<>(); for (String partitionPath : partitionPaths) { partitionPathsAbs.add(basePath + "/" + partitionPath + "/*"); } int size = 10 + RANDOM.nextInt(100); int batches = 1; Dataset<Row> totalInputRows = null; for (int j = 0; j < batches; j++) { String partitionPath = HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[j % 3]; Dataset<Row> inputRows = getRandomRows(sqlContext, size, partitionPath, false); writeRows(inputRows, writer); if (totalInputRows == null) { totalInputRows = inputRows; } else { totalInputRows = totalInputRows.union(inputRows); } } HoodieWriterCommitMessage commitMetadata = (HoodieWriterCommitMessage) writer.commit(); List<HoodieWriterCommitMessage> commitMessages = new ArrayList<>(); commitMessages.add(commitMetadata); dataSourceInternalWriter.commit(commitMessages.toArray(new HoodieWriterCommitMessage[0])); metaClient.reloadActiveTimeline(); Dataset<Row> result = HoodieClientTestUtils.read(jsc, basePath, sqlContext, metaClient.getFs(), partitionPathsAbs.toArray(new String[0])); assertOutput(totalInputRows, result, instantTime0); assertWriteStatuses(commitMessages.get(0).getWriteStatuses(), batches, size); String instantTime1 = "00" + 1; dataSourceInternalWriter = new HoodieDataSourceInternalWriter(instantTime1, cfg, STRUCT_TYPE, sqlContext.sparkSession(), hadoopConf); writer = dataSourceInternalWriter.createWriterFactory().createDataWriter(1, RANDOM.nextLong(), RANDOM.nextLong()); for (int j = 0; j < batches; j++) { String partitionPath = HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[j % 3]; Dataset<Row> inputRows = getRandomRows(sqlContext, size, partitionPath, false); writeRows(inputRows, writer); } commitMetadata = (HoodieWriterCommitMessage) writer.commit(); commitMessages = new ArrayList<>(); commitMessages.add(commitMetadata); dataSourceInternalWriter.abort(commitMessages.toArray(new HoodieWriterCommitMessage[0])); metaClient.reloadActiveTimeline(); result = HoodieClientTestUtils.read(jsc, basePath, sqlContext, metaClient.getFs(), partitionPathsAbs.toArray(new String[0])); assertOutput(totalInputRows, result, instantTime0); }
@Override public void abort(WriterCommitMessage[] messages) { LOG.error("Commit " + instantTime + " aborted "); writeClient.rollback(instantTime); writeClient.close(); }
HoodieDataSourceInternalWriter implements DataSourceWriter { @Override public void abort(WriterCommitMessage[] messages) { LOG.error("Commit " + instantTime + " aborted "); writeClient.rollback(instantTime); writeClient.close(); } }
HoodieDataSourceInternalWriter implements DataSourceWriter { @Override public void abort(WriterCommitMessage[] messages) { LOG.error("Commit " + instantTime + " aborted "); writeClient.rollback(instantTime); writeClient.close(); } HoodieDataSourceInternalWriter(String instantTime, HoodieWriteConfig writeConfig, StructType structType,
                                        SparkSession sparkSession, Configuration configuration); }
HoodieDataSourceInternalWriter implements DataSourceWriter { @Override public void abort(WriterCommitMessage[] messages) { LOG.error("Commit " + instantTime + " aborted "); writeClient.rollback(instantTime); writeClient.close(); } HoodieDataSourceInternalWriter(String instantTime, HoodieWriteConfig writeConfig, StructType structType,
                                        SparkSession sparkSession, Configuration configuration); @Override DataWriterFactory<InternalRow> createWriterFactory(); @Override boolean useCommitCoordinator(); @Override void onDataWriterCommit(WriterCommitMessage message); @Override void commit(WriterCommitMessage[] messages); @Override void abort(WriterCommitMessage[] messages); }
HoodieDataSourceInternalWriter implements DataSourceWriter { @Override public void abort(WriterCommitMessage[] messages) { LOG.error("Commit " + instantTime + " aborted "); writeClient.rollback(instantTime); writeClient.close(); } HoodieDataSourceInternalWriter(String instantTime, HoodieWriteConfig writeConfig, StructType structType,
                                        SparkSession sparkSession, Configuration configuration); @Override DataWriterFactory<InternalRow> createWriterFactory(); @Override boolean useCommitCoordinator(); @Override void onDataWriterCommit(WriterCommitMessage message); @Override void commit(WriterCommitMessage[] messages); @Override void abort(WriterCommitMessage[] messages); static final String INSTANT_TIME_OPT_KEY; }