@Override
    public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
        alluxioSinkConfig = AlluxioSinkConfig.load(config);
        alluxioSinkConfig.validate();

        // initialize FileSystem
        String alluxioMasterHost = alluxioSinkConfig.getAlluxioMasterHost();
        int alluxioMasterPort = alluxioSinkConfig.getAlluxioMasterPort();
        configuration.set(PropertyKey.MASTER_HOSTNAME, alluxioMasterHost);
        configuration.set(PropertyKey.MASTER_RPC_PORT, alluxioMasterPort);
        if (alluxioSinkConfig.getSecurityLoginUser() != null) {
            configuration.set(PropertyKey.SECURITY_LOGIN_USERNAME, alluxioSinkConfig.getSecurityLoginUser());
        }
        fileSystem = FileSystem.Factory.create(configuration);

        // initialize alluxio dirs
        String alluxioDir = alluxioSinkConfig.getAlluxioDir();
        fileDirPath = alluxioDir.startsWith("/") ? alluxioDir : "/" + alluxioDir;
        tmpFileDirPath = fileDirPath + "/tmp";

        AlluxioURI alluxioDirPath = new AlluxioURI(fileDirPath);
        if (!fileSystem.exists(alluxioDirPath)) {
            fileSystem.createDirectory(alluxioDirPath);
        }

        AlluxioURI tmpAlluxioDirPath = new AlluxioURI(tmpFileDirPath);
        if (!fileSystem.exists(tmpAlluxioDirPath)) {
            fileSystem.createDirectory(tmpAlluxioDirPath);
        }

        recordsNum = 0;
        recordsToAck = Lists.newArrayList();
        tmpFilePath = "";
        alluxioState = AlluxioState.WRITE_STARTED;

        lastRotationTime = System.currentTimeMillis();
        rotationRecordsNum = alluxioSinkConfig.getRotationRecords();
        rotationInterval =  alluxioSinkConfig.getRotationInterval();
    }
@Test
    public void openTest() throws Exception {
        map.put("filePrefix", "TopicA");
        map.put("fileExtension", ".txt");
        map.put("lineSeparator", "\n");
        map.put("rotationRecords", "100");

        String alluxioDir = "/pulsar";

        sink = new AlluxioSink();
        sink.open(map, mockSinkContext);

        FileSystem client = cluster.getClient();

        AlluxioURI alluxioURI = new AlluxioURI(alluxioDir);
        Assert.assertTrue(client.exists(alluxioURI));

        String alluxioTmpDir = FilenameUtils.concat(alluxioDir, "tmp");
        AlluxioURI alluxioTmpURI = new AlluxioURI(alluxioTmpDir);
        Assert.assertTrue(client.exists(alluxioTmpURI));

        sink.close();
    }