morningman commented on code in PR #51539:
URL: https://github.com/apache/doris/pull/51539#discussion_r2133853455
##########
fe/fe-core/src/main/java/org/apache/doris/fs/remote/dfs/DFSFileSystem.java:
##########
@@ -55,22 +55,26 @@
import java.nio.file.Paths;
import java.util.Comparator;
import java.util.List;
-import java.util.Map;
public class DFSFileSystem extends RemoteFileSystem {
public static final String PROP_ALLOW_FALLBACK_TO_SIMPLE_AUTH =
"ipc.client.fallback-to-simple-auth-allowed";
private static final Logger LOG =
LogManager.getLogger(DFSFileSystem.class);
private HDFSFileOperations operations = null;
private HadoopAuthenticator authenticator = null;
+ private HdfsCompatibleProperties hdfsProperties;
- public DFSFileSystem(Map<String, String> properties) {
- this(StorageBackend.StorageType.HDFS, properties);
+ public DFSFileSystem(HdfsCompatibleProperties hdfsProperties) {
+ super(StorageBackend.StorageType.HDFS.name(),
StorageBackend.StorageType.HDFS);
+ this.properties.putAll(hdfsProperties.getOrigProps());
+ this.storageProperties = hdfsProperties;
Review Comment:
Why need both `storageProperties` and `hdfsProperties`
##########
fe/fe-core/src/main/java/org/apache/doris/fs/remote/S3FileSystem.java:
##########
@@ -20,36 +20,36 @@
import org.apache.doris.analysis.StorageBackend;
import org.apache.doris.backup.Status;
import org.apache.doris.common.UserException;
-import org.apache.doris.common.security.authentication.AuthenticationConfig;
import org.apache.doris.common.security.authentication.HadoopAuthenticator;
-import org.apache.doris.datasource.property.PropertyConverter;
-import org.apache.doris.datasource.property.constants.S3Properties;
+import org.apache.doris.common.util.S3URI;
+import
org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties;
import org.apache.doris.fs.obj.S3ObjStorage;
-import org.apache.doris.fs.remote.dfs.DFSFileSystem;
-import com.amazonaws.services.s3.model.AmazonS3Exception;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Strings;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import software.amazon.awssdk.services.s3.S3Client;
-import java.io.FileNotFoundException;
-import java.io.IOException;
+import java.util.HashSet;
import java.util.List;
-import java.util.Map;
+import java.util.Set;
public class S3FileSystem extends ObjFileSystem {
private static final Logger LOG = LogManager.getLogger(S3FileSystem.class);
private HadoopAuthenticator authenticator = null;
+ private AbstractS3CompatibleProperties s3Properties;
- public S3FileSystem(Map<String, String> properties) {
- super(StorageBackend.StorageType.S3.name(),
StorageBackend.StorageType.S3, new S3ObjStorage(properties));
+
+ public S3FileSystem(AbstractS3CompatibleProperties s3Properties) {
+
+ super(StorageBackend.StorageType.S3.name(),
StorageBackend.StorageType.S3,
+ new S3ObjStorage(s3Properties));
+ this.s3Properties = s3Properties;
Review Comment:
Why need 2 properties? `s3Properties` and `storageProperties`?
##########
fe/fe-core/src/main/java/org/apache/doris/fs/obj/ObjStorage.java:
##########
@@ -34,7 +34,7 @@
public interface ObjStorage<C> {
// CHUNK_SIZE for multi part upload
- public static final int CHUNK_SIZE = 5 * 1024 * 1024;
+ int CHUNK_SIZE = 5 * 1024 * 1024;
Review Comment:
why removing `public static final`?
##########
fe/fe-core/src/main/java/org/apache/doris/fs/remote/S3FileSystem.java:
##########
@@ -59,107 +59,44 @@ public S3FileSystem(S3ObjStorage storage) {
}
private void initFsProperties() {
- this.properties.putAll(((S3ObjStorage) objStorage).getProperties());
+ this.properties.putAll(storageProperties.getOrigProps());
}
+
@Override
protected FileSystem nativeFileSystem(String remotePath) throws
UserException {
- //todo Extracting a common method to achieve logic reuse
- if (closed.get()) {
- throw new UserException("FileSystem is closed.");
- }
- if (dfsFileSystem == null) {
- synchronized (this) {
- if (closed.get()) {
- throw new UserException("FileSystem is closed.");
- }
- if (dfsFileSystem == null) {
- Configuration conf =
DFSFileSystem.getHdfsConf(ifNotSetFallbackToSimpleAuth());
- System.setProperty("com.amazonaws.services.s3.enableV4",
"true");
- // the entry value in properties may be null, and
-
PropertyConverter.convertToHadoopFSProperties(properties).entrySet().stream()
- .filter(entry -> entry.getKey() != null &&
entry.getValue() != null)
- .forEach(entry -> conf.set(entry.getKey(),
entry.getValue()));
- // S3 does not support Kerberos authentication,
- // so here we create a simple authentication
- AuthenticationConfig authConfig =
AuthenticationConfig.getSimpleAuthenticationConfig(conf);
- HadoopAuthenticator authenticator =
HadoopAuthenticator.getHadoopAuthenticator(authConfig);
- try {
- dfsFileSystem = authenticator.doAs(() -> {
- try {
- return FileSystem.get(new
Path(remotePath).toUri(), conf);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- });
- this.authenticator = authenticator;
- RemoteFSPhantomManager.registerPhantomReference(this);
- } catch (Exception e) {
- throw new UserException("Failed to get S3 FileSystem
for " + e.getMessage(), e);
- }
- }
- }
- }
- return dfsFileSystem;
+ throw new UserException("S3 does not support native file system");
}
// broker file pattern glob is too complex, so we use hadoop directly
Review Comment:
modify the comment
##########
fe/fe-core/src/main/java/org/apache/doris/fs/remote/dfs/DFSFileSystem.java:
##########
@@ -85,12 +89,11 @@ public FileSystem nativeFileSystem(String remotePath)
throws UserException {
throw new UserException("FileSystem is closed.");
}
if (dfsFileSystem == null) {
- Configuration conf =
getHdfsConf(ifNotSetFallbackToSimpleAuth());
- for (Map.Entry<String, String> propEntry :
properties.entrySet()) {
- conf.set(propEntry.getKey(), propEntry.getValue());
- }
- AuthenticationConfig authConfig =
AuthenticationConfig.getKerberosConfig(conf);
- authenticator =
HadoopAuthenticator.getHadoopAuthenticator(authConfig);
+ Configuration conf =
hdfsProperties.getHadoopConfiguration();
+ // TODO: Temporarily disable the HDFS file system cache to
prevent instances from being closed by
+ // each other in V1. This line can be removed once V1 and
V2 are unified.
Review Comment:
There is anything else we need to do to unify v1 and v2?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
