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.carbondata.core.locks;
  
  import java.lang.reflect.Constructor;
  import java.lang.reflect.InvocationTargetException;
  import java.lang.reflect.Modifier;
  
  import org.apache.carbondata.common.logging.LogServiceFactory;
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.datastore.impl.FileFactory;
  import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
  import org.apache.carbondata.core.util.CarbonProperties;
  
  import org.apache.log4j.Logger;
  
  /**
   * This class is a Lock factory class which is used to provide lock objects.
   * Using this lock object client can request the lock and unlock.
   */
  public class CarbonLockFactory {
  
    /**
     * Attribute for LOGGER
     */
    private static final Logger LOGGER =
        LogServiceFactory.getLogService(CarbonLockFactory.class.getName());
    /**
     * lockTypeConfigured to check if zookeeper feature is enabled or not for carbon.
     */
    private static String lockTypeConfigured;
  
    private static String lockPath = CarbonProperties.getInstance()
        .getProperty(CarbonCommonConstants.LOCK_PATH, CarbonCommonConstants.LOCK_PATH_DEFAULT)
        .toLowerCase();
  
    private static Constructor lockConstructor;
  
    static {
      CarbonLockFactory.getLockTypeConfigured();
    }
  
    /**
     * This method will determine the lock type.
     *
     * @param absoluteTableIdentifier
     * @param lockFile
     * @return
     */
    public static ICarbonLock getCarbonLockObj(AbsoluteTableIdentifier absoluteTableIdentifier,
        String lockFile) {
      String absoluteLockPath;
      if (lockPath.isEmpty()) {
        absoluteLockPath = absoluteTableIdentifier.getTablePath();
      } else {
        absoluteLockPath =
            getLockPath(absoluteTableIdentifier.getCarbonTableIdentifier().getTableId());
      }
      FileFactory.FileType fileType = FileFactory.getFileType(absoluteLockPath);
      if (lockTypeConfigured.equals(CarbonCommonConstants.CARBON_LOCK_TYPE_CUSTOM)) {
        return newCustomLock(absoluteLockPath, lockFile);
      } else if (lockTypeConfigured.equals(CarbonCommonConstants.CARBON_LOCK_TYPE_ZOOKEEPER)) {
        return new ZooKeeperLocking(absoluteLockPath, lockFile);
      } else if (fileType == FileFactory.FileType.S3) {
        lockTypeConfigured = CarbonCommonConstants.CARBON_LOCK_TYPE_S3;
        return new S3FileLock(absoluteLockPath,
                  lockFile);
      } else if (fileType == FileFactory.FileType.HDFS || fileType == FileFactory.FileType.VIEWFS) {
        lockTypeConfigured = CarbonCommonConstants.CARBON_LOCK_TYPE_HDFS;
        return new HdfsFileLock(absoluteLockPath, lockFile);
      } else if (fileType == FileFactory.FileType.ALLUXIO) {
        lockTypeConfigured = CarbonCommonConstants.CARBON_LOCK_TYPE_ALLUXIO;
        return new AlluxioFileLock(absoluteLockPath, lockFile);
      } else {
        lockTypeConfigured = CarbonCommonConstants.CARBON_LOCK_TYPE_LOCAL;
        return new LocalFileLock(absoluteLockPath, lockFile);
      }
    }
  
    /**
     *
     * @param locFileLocation
     * @param lockFile
     * @return carbon lock
     */
    public static ICarbonLock getSystemLevelCarbonLockObj(String locFileLocation, String lockFile) {
      String lockFileLocation;
      if (lockPath.isEmpty()) {
        lockFileLocation = locFileLocation;
      } else {
        lockFileLocation = getLockPath("1");
      }
      switch (lockTypeConfigured) {
        case CarbonCommonConstants.CARBON_LOCK_TYPE_CUSTOM:
          return newCustomLock(lockFileLocation, lockFile);
        case CarbonCommonConstants.CARBON_LOCK_TYPE_LOCAL:
          return new LocalFileLock(lockFileLocation, lockFile);
        case CarbonCommonConstants.CARBON_LOCK_TYPE_ZOOKEEPER:
          return new ZooKeeperLocking(lockFileLocation, lockFile);
        case CarbonCommonConstants.CARBON_LOCK_TYPE_HDFS:
          return new HdfsFileLock(lockFileLocation, lockFile);
        case CarbonCommonConstants.CARBON_LOCK_TYPE_S3:
          return new S3FileLock(lockFileLocation, lockFile);
        case CarbonCommonConstants.CARBON_LOCK_TYPE_ALLUXIO:
          return new AlluxioFileLock(lockFileLocation, lockFile);
        default:
          throw new UnsupportedOperationException("Not supported the lock type");
      }
    }
  
    /**
     * This method will set the zookeeper status whether zookeeper to be used for locking or not.
     */
    private static void getLockTypeConfigured() {
      lockTypeConfigured = CarbonProperties.getInstance()
          .getProperty(CarbonCommonConstants.LOCK_TYPE, CarbonCommonConstants.LOCK_TYPE_DEFAULT)
          .toUpperCase();
      LOGGER.info("Configured lock type is: " + lockTypeConfigured);
      String lockClassName =
          CarbonProperties.getInstance().getProperty(CarbonCommonConstants.LOCK_CLASS);
      if (lockClassName == null) {
        return;
      }
      CarbonLockFactory.lockConstructor = getCustomLockConstructor(lockClassName);
    }
  
    public static String getLockPath(String tableId) {
      return lockPath + CarbonCommonConstants.FILE_SEPARATOR + tableId;
    }
  
    private static ICarbonLock newCustomLock(final String lockFileLocation, final String lockFile) {
      if (lockConstructor == null) {
        throw new IllegalArgumentException(
            "Carbon property [" + CarbonCommonConstants.LOCK_CLASS + "] is not set.");
      }
      try {
        return (ICarbonLock) lockConstructor.newInstance(lockFileLocation, lockFile);
      } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
        throw new RuntimeException(e);
      }
    }
  
    private static Constructor<?> getCustomLockConstructor(final String lockClassName) {
      final Class<?> lockClass;
      try {
        lockClass = CarbonLockFactory.class.getClassLoader().loadClass(lockClassName);
      } catch (ClassNotFoundException e) {
        throw new IllegalArgumentException("The class [" + lockClassName + "] is not found.");
      }
      if (!ICarbonLock.class.isAssignableFrom(lockClass)) {
        throw new IllegalArgumentException(
            "The class [" + lockClassName + "] is not an ICarbonLock class.");
      }
      if (Modifier.isAbstract(lockClass.getModifiers())) {
        throw new IllegalArgumentException(
            "The class [" + lockClassName + "] can not be initialized.");
      }
      if (!Modifier.isPublic(lockClass.getModifiers())) {
        throw new IllegalArgumentException(
            "The class [" + lockClassName + "] is not a public class.");
      }
      final Constructor<?> lockConstructor;
      try {
        lockConstructor = lockClass.getConstructor(String.class, String.class);
      } catch (NoSuchMethodException e) {
        throw new IllegalArgumentException(
            "The class [" + lockClassName + "] do not have the constructor(String, String).", e
        );
      }
      if (!Modifier.isPublic(lockClass.getModifiers())) {
        throw new IllegalArgumentException(
            "The constructor [" + lockConstructor + "] is not a public constructor.");
      }
      return lockConstructor;
    }
  }
  