class PaloLoadTask extends TimerTask {

        private static final String DF = "yyyy-MM-dd-HH-mm-ss";
        private final SafeDateFormat dateFormat = new SafeDateFormat(DF);

        /**
         * There exist two running palo load task corresponds to two stores,
         * `SchemaStore -> SchemaLoadTask` and `GraphStore -> GraphLoadTask`,
         * the load task just handle with it's own subdirectory files,
         * like: `property_keys, vertex_labels ...` for SchemaLoadTask,
         * and: `vertices, edges ...` for GraphLoadTask
         * these subdirectory called validSubDirs.
         */
        private final List<String> tableDirs;
        /**
         *          session1-part1.txt  session2-part1.txt  ...
         * vertices 128m                125m                ...
         * edges    256m                250m                ...
         * ...      ...                 ...                 ...
         */
        private List<PaloFile> lastPaloFiles;
        private final PaloHttpClient client;

        public PaloLoadTask(List<String> tableDirs) {
            this.tableDirs = tableDirs;
            this.lastPaloFiles = null;
            this.client = new PaloHttpClient(config(), database());
        }

        /**
         * TODO: Need to be optimized
         */
        public void join() {
            Integer interval = config().get(PaloOptions.PALO_POLL_INTERVAL);
            while (this.lastPaloFiles != null &&
                   !this.lastPaloFiles.isEmpty()) {
                try {
                    TimeUnit.SECONDS.sleep(interval);
                } catch (InterruptedException e) {
                    throw new BackendException(e);
                }
            }
        }

        @Override
        public void run() {<FILL_FUNCTION_BODY>}

        private void tryLoadBatch(List<PaloFile> files) {
            PaloFile file = this.peekFile(files);
            // Load the first file when stopped inserting data
            this.loadThenDelete(file);
            files.remove(file);
        }

        private PaloFile peekFile(List<PaloFile> files) {
            assert !files.isEmpty();
            long limitSize = PaloFile.limitSize(config());
            // Load the file which exceed limit size in priority
            for (PaloFile file : files) {
                long fileSize = file.length();
                if (fileSize >= limitSize) {
                    return file;
                }
            }
            // Load the oldest file(files are sorted by updated time)
            return files.get(0);
        }

        private void loadThenDelete(PaloFile file) {
            // Parse session id from file name
            int sessionId = file.sessionId();
            LOG.info("Ready to load one batch from file: {}", file);
            // Get write lock because will delete file
            Lock lock = PaloSessions.this.locks.get(sessionId).writeLock();
            lock.lock();
            try {
                String table = file.table();
                String data = file.readAsString();
                String label = this.formatLabel(table);
                this.client.bulkLoadAsync(table, data, label);
                // Force delete file
                file.forceDelete();
            } finally {
                lock.unlock();
            }
        }

        private String formatLabel(String table) {
            return table + "-" + this.dateFormat.format(new Date());
        }
    }

            LOG.debug("The Load task:{} ready to run", PaloSessions.this);
            // Scan the directory to get all file size
            String path = config().get(PaloOptions.PALO_TEMP_DIR);
            List<PaloFile> paloFiles = PaloFile.scan(path, this.tableDirs);
            // Do nothing if there is no file at present
            if (paloFiles.isEmpty()) {
                return;
            }
            // Try to load one batch if last time and this time have some files
            if (this.lastPaloFiles != null) {
                this.tryLoadBatch(paloFiles);
            }
            // Update the last palo files
            this.lastPaloFiles = paloFiles;
        
<methods>public void <init>(HugeConfig, java.lang.String, java.lang.String) ,public HugeConfig config() ,public void createDatabase() ,public java.lang.String database() ,public void dropDatabase() ,public java.lang.String escapedDatabase() ,public boolean existsDatabase() ,public boolean existsTable(java.lang.String) ,public synchronized void open() throws java.lang.Exception,public void resetConnections() ,public org.apache.hugegraph.backend.store.mysql.MysqlSessions.Session session() <variables>private static final int DROP_DB_TIMEOUT,private static final java.lang.String JDBC_PREFIX,private static final Logger LOG,private HugeConfig config,private java.lang.String database,private volatile boolean opened