Skip to content

[HUDI-5489] Flink offline compactor throws exception in service mode #7588

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danny0405 merged 1 commit into from
Jan 3, 2023
Merged

[HUDI-5489] Flink offline compactor throws exception in service mode #7588

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 4 additions & 10 deletions ...ce/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,12 +76,10 @@ public HoodieFlinkClusteringJob(AsyncClusteringService service) {
}

public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

FlinkClusteringConfig cfg = getFlinkClusteringConfig(args);
Configuration conf = FlinkClusteringConfig.toFlinkConfig(cfg);

AsyncClusteringService service = new AsyncClusteringService(cfg, conf, env);
AsyncClusteringService service = new AsyncClusteringService(cfg, conf);

new HoodieFlinkClusteringJob(service).start(cfg.serviceMode);
}
Expand Down Expand Up @@ -165,20 +163,14 @@ public static class AsyncClusteringService extends HoodieAsyncTableService {
*/
private final HoodieFlinkTable<?> table;

/**
* Flink Execution Environment.
*/
private final StreamExecutionEnvironment env;

/**
* Executor Service.
*/
private final ExecutorService executor;

public AsyncClusteringService(FlinkClusteringConfig cfg, Configuration conf, StreamExecutionEnvironment env) throws Exception {
public AsyncClusteringService(FlinkClusteringConfig cfg, Configuration conf) throws Exception {
this.cfg = cfg;
this.conf = conf;
this.env = env;
this.executor = Executors.newFixedThreadPool(1);

// create metaClient
Expand Down Expand Up @@ -338,6 +330,8 @@ private void cluster() throws Exception {
final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
final RowType rowType = (RowType) rowDataType.getLogicalType();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// setup configuration
long ckpTimeout = env.getCheckpointConfig().getCheckpointTimeout();
conf.setLong(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT, ckpTimeout);
Expand Down
13 changes: 3 additions & 10 deletions ...atasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,12 +69,10 @@ public HoodieFlinkCompactor(AsyncCompactionService service) {
}

public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

FlinkCompactionConfig cfg = getFlinkCompactionConfig(args);
Configuration conf = FlinkCompactionConfig.toFlinkConfig(cfg);

AsyncCompactionService service = new AsyncCompactionService(cfg, conf, env);
AsyncCompactionService service = new AsyncCompactionService(cfg, conf);

new HoodieFlinkCompactor(service).start(cfg.serviceMode);
}
Expand Down Expand Up @@ -157,20 +155,14 @@ public static class AsyncCompactionService extends HoodieAsyncTableService {
*/
private final HoodieFlinkTable<?> table;

/**
* Flink Execution Environment.
*/
private final StreamExecutionEnvironment env;

/**
* Executor Service.
*/
private final ExecutorService executor;

public AsyncCompactionService(FlinkCompactionConfig cfg, Configuration conf, StreamExecutionEnvironment env) throws Exception {
public AsyncCompactionService(FlinkCompactionConfig cfg, Configuration conf) throws Exception {
this.cfg = cfg;
this.conf = conf;
this.env = env;
this.executor = Executors.newFixedThreadPool(1);

// create metaClient
Expand Down Expand Up @@ -304,6 +296,7 @@ private void compact() throws Exception {
}
table.getMetaClient().reloadActiveTimeline();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(new CompactionPlanSourceFunction(compactionPlans))
.name("compaction_source")
.uid("uid_compaction_source")
Expand Down
3 changes: 1 addition & 2 deletions ...ce/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,14 +208,13 @@ public void testHoodieFlinkClusteringService() throws Exception {
TimeUnit.SECONDS.sleep(3);

// Make configuration and setAvroSchema.
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
FlinkClusteringConfig cfg = new FlinkClusteringConfig();
cfg.path = tempFile.getAbsolutePath();
cfg.minClusteringIntervalSeconds = 3;
cfg.schedule = true;
Configuration conf = FlinkClusteringConfig.toFlinkConfig(cfg);

HoodieFlinkClusteringJob.AsyncClusteringService asyncClusteringService = new HoodieFlinkClusteringJob.AsyncClusteringService(cfg, conf, env);
HoodieFlinkClusteringJob.AsyncClusteringService asyncClusteringService = new HoodieFlinkClusteringJob.AsyncClusteringService(cfg, conf);
asyncClusteringService.start(null);

// wait for the asynchronous commit to finish
Expand Down
3 changes: 1 addition & 2 deletions ...rce/hudi-flink/src/test/java/org/apache/hudi/sink/compact/ITTestHoodieFlinkCompactor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,6 @@ public void testHoodieFlinkCompactorService(boolean enableChangelog) throws Exce
tableEnv.executeSql(TestSQL.UPDATE_INSERT_T1).await();

// Make configuration and setAvroSchema.
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
FlinkCompactionConfig cfg = new FlinkCompactionConfig();
cfg.path = tempFile.getAbsolutePath();
cfg.minCompactionIntervalSeconds = 3;
Expand All @@ -200,7 +199,7 @@ public void testHoodieFlinkCompactorService(boolean enableChangelog) throws Exce
conf.setString(FlinkOptions.TABLE_TYPE.key(), "MERGE_ON_READ");
conf.setInteger(FlinkOptions.COMPACTION_TASKS.key(), FlinkMiniCluster.DEFAULT_PARALLELISM);

HoodieFlinkCompactor.AsyncCompactionService asyncCompactionService = new HoodieFlinkCompactor.AsyncCompactionService(cfg, conf, env);
HoodieFlinkCompactor.AsyncCompactionService asyncCompactionService = new HoodieFlinkCompactor.AsyncCompactionService(cfg, conf);
asyncCompactionService.start(null);

// wait for the asynchronous commit to finish
Expand Down