package com.primeton.damp.core.config;

import com.primeton.damp.principal.HadoopPrincipal;
import com.primeton.damp.principal.HadoopPrincipalFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;
import java.util.Optional;

/**
 * <pre>
 *
 * Created by zhaopx.
 * User: zhaopx
 * Date: 2021/4/2
 * Time: 15:54
 *
 * </pre>
 *
 * @author zhaopx
 */
@Configuration
@Slf4j
public class HadoopConfigurer {


    /**
     * hdfs
     */
    @Autowired
    HadoopProperties hadoopProperties;

    @Bean
    public FileSystem fs() throws IOException {
        String type = Optional.ofNullable(hadoopProperties.getHdfsType()).orElse("local");
        if("local".equalsIgnoreCase(type)) {
            // local
            log.info("use local file system");
            return new RawLocalFileSystem();
        } else if("cluster".equalsIgnoreCase(type)) {
            log.info("use hdfs filesystem");
            final HadoopPrincipal hadoopPrincipal = HadoopPrincipalFactory.getHadoopPrincipal(hadoopProperties.getTenentName(), hadoopProperties.getUserKeytab(), hadoopProperties.getKrb5File());
            final org.apache.hadoop.conf.Configuration conf = hadoopPrincipal.getConf(hadoopProperties.getCoreSite(), hadoopProperties.getHdfsSite());
            return FileSystem.get(conf);
        } else {
            throw new IllegalStateException("unkown hadoop type: " + type);
        }
    }
}