Merge pull request #76 from arenadata/feature/ADBDEV-4980

ADBDEV-4980: Implement reload and reloadAll methods for profiles

server/pxf-api/src/main/java/org/greenplum/pxf/api/model/BasePlugin.java

@@ -34,6 +34,14 @@ public void setRequestContext(RequestContext context) {
     public void afterPropertiesSet() {
     }
 
+    @Override
+    public void reloadAll() {
+    }
+
+    @Override
+    public void reload(String server) {
+    }
+
     /**
      * When DEBUG mode is enabled, logs the total number of rows read, the
      * amount of time it took to read the file, and the average read speed

server/pxf-api/src/main/java/org/greenplum/pxf/api/model/Plugin.java

@@ -16,4 +16,8 @@ public interface Plugin {
      * Invoked after the {@code RequestContext} has been bound
      */
     void afterPropertiesSet();
+
+    void reloadAll();
+
+    void reload(String server);
 }

server/pxf-jdbc/src/main/java/org/greenplum/pxf/plugins/jdbc/JdbcBasePlugin.java

@@ -22,6 +22,7 @@
 import io.arenadata.security.encryption.client.service.DecryptClient;
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
+import org.greenplum.pxf.api.error.PxfRuntimeException;
 import org.greenplum.pxf.api.model.BasePlugin;
 import org.greenplum.pxf.api.model.RequestContext;
 import org.greenplum.pxf.api.security.SecureLogin;
@@ -43,9 +44,9 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.stream.Collectors;
-import java.util.Objects;
 
 import static org.greenplum.pxf.api.security.SecureLogin.CONFIG_KEY_SERVICE_USER_IMPERSONATION;
 
@@ -100,6 +101,7 @@ public class JdbcBasePlugin extends BasePlugin {
     private static final String MYSQL_DRIVER_PREFIX = "com.mysql.";
     private static final String JDBC_DATE_WIDE_RANGE = "jdbc.date.wideRange";
     private static final String JDBC_DATE_WIDE_RANGE_LEGACY = "jdbc.date.wide-range";
+
     private enum TransactionIsolation {
         READ_UNCOMMITTED(1),
         READ_COMMITTED(2),
@@ -493,6 +495,26 @@ public static void closeStatementAndConnection(Statement statement) throws SQLEx
         }
     }
 
+    @Override
+    public void reloadAll() {
+        if (Objects.nonNull(connectionManager)) {
+            connectionManager.reloadCache();
+        } else {
+            throw new PxfRuntimeException("Failed to reload profile. Connection manager is null.");
+        }
+    }
+
+    @Override
+    public void reload(String server) {
+        if (Objects.isNull(connectionManager)) {
+            throw new PxfRuntimeException("Failed to reload profile. Connection manager is null.");
+        } else if (StringUtils.isBlank(server)) {
+            throw new PxfRuntimeException("Failed to reload profile. Parameter server is blank.");
+        } else {
+            connectionManager.reloadCacheIf(poolDescriptor -> poolDescriptor.getServer().equals(server));
+        }
+    }
+
     /**
      * For a Kerberized Hive JDBC connection, it creates a connection as the loginUser.
      * Otherwise, it returns a new connection.

server/pxf-jdbc/src/main/java/org/greenplum/pxf/plugins/jdbc/utils/ConnectionManager.java

@@ -11,6 +11,7 @@
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
 import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang.StringUtils;
 import org.greenplum.pxf.plugins.jdbc.PxfJdbcProperties;
 import org.slf4j.Logger;
@@ -27,11 +28,14 @@
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 /**
  * Responsible for obtaining and maintaining JDBC connections to databases. If configured for a given server,
  * uses Hikari Connection Pool to pool database connections.
  */
+@Slf4j
 @Component
 public class ConnectionManager {
 
@@ -78,6 +82,23 @@ public ConnectionManager(DataSourceFactory factory, Ticker ticker, PxfJdbcProper
                 .build(CacheLoader.from(factory::createDataSource));
     }
 
+    public void reloadCache() {
+        log.info("Invalidate cache of all pool descriptors");
+        dataSources.invalidateAll();
+        cleanCache();
+    }
+
+    public void reloadCacheIf(Predicate<PoolDescriptor> poolDescriptorFilter) {
+        dataSources.asMap().keySet().stream()
+                .filter(poolDescriptorFilter)
+                .collect(Collectors.toSet())
+                .forEach(poolDescriptor -> {
+                    log.info("Invalidate cache of the pool descriptor {}", poolDescriptor);
+                    dataSources.invalidate(poolDescriptor);
+                });
+        cleanCache();
+    }
+
     /**
      * Explicitly runs cache maintenance operations.
      */

server/pxf-jdbc/src/main/java/org/greenplum/pxf/plugins/jdbc/utils/PoolDescriptor.java

@@ -90,7 +90,8 @@ public int hashCode() {
     @Override
     public String toString() {
         return "PoolDescriptor{" +
-                "jdbcUrl=" + jdbcUrl +
+                "server=" + server +
+                ", jdbcUrl=" + jdbcUrl +
                 ", user=" + user +
                 ", password=" + ConnectionManager.maskPassword(password) +
                 ", connectionConfig=" + connectionConfig +