lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2568] Simplify the view storage config properties (#3815)

Browse Source
This commit is contained in:
Danny Chan
2021-10-18 14:42:33 +08:00
committed by GitHub
parent 5276850415
commit 3025f4d796
2 changed files with 11 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
hudi-flink/src/main/java/org/apache/hudi/sink/utils/TimeWait.java
View File

@@ -35,14 +35,14 @@ public class TimeWait {
private final long timeout; // timeout in SECONDS private final long timeout; // timeout in SECONDS
private final long interval; // interval in MILLISECONDS private final long interval; // interval in MILLISECONDS
private final String action; // action to report error message private final String action; // action to report error message
private final boolean throwsE; // whether to throw when timeout private final boolean throwsT; // whether to throw when timeout
private long waitingTime = 0L; private long waitingTime = 0L;
private TimeWait(long timeout, long interval, String action, boolean throwsE) { private TimeWait(long timeout, long interval, String action, boolean throwsT) {
this.timeout = timeout; this.timeout = timeout;
this.interval = interval; this.interval = interval;
this.action = action; this.action = action;
this.throwsE = throwsE; this.throwsT = throwsT;
} }
public static Builder builder() { public static Builder builder() {
@@ -58,7 +58,7 @@ public class TimeWait {
try { try {
if (waitingTime > timeout) { if (waitingTime > timeout) {
final String msg = "Timeout(" + waitingTime + "ms) while waiting for " + action; final String msg = "Timeout(" + waitingTime + "ms) while waiting for " + action;
if (this.throwsE) { if (this.throwsT) {
throw new HoodieException(msg); throw new HoodieException(msg);
} else { } else {
LOG.warn(msg); LOG.warn(msg);

8
hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
View File

@@ -392,7 +392,13 @@ public class StreamerUtil {
public static HoodieFlinkWriteClient createWriteClient(Configuration conf) throws IOException { public static HoodieFlinkWriteClient createWriteClient(Configuration conf) throws IOException {
HoodieWriteConfig writeConfig = getHoodieClientConfig(conf, true, false); HoodieWriteConfig writeConfig = getHoodieClientConfig(conf, true, false);
// create the filesystem view storage properties for client // create the filesystem view storage properties for client
ViewStorageProperties.createProperties(conf.getString(FlinkOptions.PATH), writeConfig.getViewStorageConfig()); FileSystemViewStorageConfig viewStorageConfig = writeConfig.getViewStorageConfig();
// rebuild the view storage config with simplified options.
FileSystemViewStorageConfig rebuilt = FileSystemViewStorageConfig.newBuilder()
.withStorageType(viewStorageConfig.getStorageType())
.withRemoteServerHost(viewStorageConfig.getRemoteViewServerHost())
.withRemoteServerPort(viewStorageConfig.getRemoteViewServerPort()).build();
ViewStorageProperties.createProperties(conf.getString(FlinkOptions.PATH), rebuilt);
return new HoodieFlinkWriteClient<>(HoodieFlinkEngineContext.DEFAULT, writeConfig); return new HoodieFlinkWriteClient<>(HoodieFlinkEngineContext.DEFAULT, writeConfig);
} }