content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.spark.network.util;
  
  import java.io.File;
  import java.io.IOException;
  import java.nio.charset.StandardCharsets;
  
  import com.fasterxml.jackson.annotation.JsonCreator;
  import com.fasterxml.jackson.annotation.JsonProperty;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import org.fusesource.leveldbjni.JniDBFactory;
  import org.fusesource.leveldbjni.internal.NativeDB;
  import org.iq80.leveldb.DB;
  import org.iq80.leveldb.Options;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * LevelDB utility class available in the network package.
   */
  public class LevelDBProvider {
    private static final Logger logger = LoggerFactory.getLogger(LevelDBProvider.class);
  
    public static DB initLevelDB(File dbFile, StoreVersion version, ObjectMapper mapper) throws
        IOException {
      DB tmpDb = null;
      if (dbFile != null) {
        Options options = new Options();
        options.createIfMissing(false);
        options.logger(new LevelDBLogger());
        try {
          tmpDb = JniDBFactory.factory.open(dbFile, options);
        } catch (NativeDB.DBException e) {
          if (e.isNotFound() || e.getMessage().contains(" does not exist ")) {
            logger.info("Creating state database at " + dbFile);
            options.createIfMissing(true);
            try {
              tmpDb = JniDBFactory.factory.open(dbFile, options);
            } catch (NativeDB.DBException dbExc) {
              throw new IOException("Unable to create state store", dbExc);
            }
          } else {
            // the leveldb file seems to be corrupt somehow.  Lets just blow it away and create a new
            // one, so we can keep processing new apps
            logger.error("error opening leveldb file {}.  Creating new file, will not be able to " +
                "recover state for existing applications", dbFile, e);
            if (dbFile.isDirectory()) {
              for (File f : dbFile.listFiles()) {
                if (!f.delete()) {
                  logger.warn("error deleting {}", f.getPath());
                }
              }
            }
            if (!dbFile.delete()) {
              logger.warn("error deleting {}", dbFile.getPath());
            }
            options.createIfMissing(true);
            try {
              tmpDb = JniDBFactory.factory.open(dbFile, options);
            } catch (NativeDB.DBException dbExc) {
              throw new IOException("Unable to create state store", dbExc);
            }
  
          }
        }
        // if there is a version mismatch, we throw an exception, which means the service is unusable
        checkVersion(tmpDb, version, mapper);
      }
      return tmpDb;
    }
  
    private static class LevelDBLogger implements org.iq80.leveldb.Logger {
      private static final Logger LOG = LoggerFactory.getLogger(LevelDBLogger.class);
  
      @Override
      public void log(String message) {
        LOG.info(message);
      }
    }
  
    /**
     * Simple major.minor versioning scheme.  Any incompatible changes should be across major
     * versions.  Minor version differences are allowed -- meaning we should be able to read
     * dbs that are either earlier *or* later on the minor version.
     */
    public static void checkVersion(DB db, StoreVersion newversion, ObjectMapper mapper) throws
        IOException {
      byte[] bytes = db.get(StoreVersion.KEY);
      if (bytes == null) {
        storeVersion(db, newversion, mapper);
      } else {
        StoreVersion version = mapper.readValue(bytes, StoreVersion.class);
        if (version.major != newversion.major) {
          throw new IOException("cannot read state DB with version " + version + ", incompatible " +
              "with current version " + newversion);
        }
        storeVersion(db, newversion, mapper);
      }
    }
  
    public static void storeVersion(DB db, StoreVersion version, ObjectMapper mapper)
        throws IOException {
      db.put(StoreVersion.KEY, mapper.writeValueAsBytes(version));
    }
  
    public static class StoreVersion {
  
      static final byte[] KEY = "StoreVersion".getBytes(StandardCharsets.UTF_8);
  
      public final int major;
      public final int minor;
  
      @JsonCreator
      public StoreVersion(@JsonProperty("major") int major, @JsonProperty("minor") int minor) {
        this.major = major;
        this.minor = minor;
      }
  
      @Override
      public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
  
        StoreVersion that = (StoreVersion) o;
  
        return major == that.major && minor == that.minor;
      }
  
      @Override
      public int hashCode() {
        int result = major;
        result = 31 * result + minor;
        return result;
      }
    }
  }
  