IGNITE-23113 [ducktests] Add thin JDBC support and LOB ducktest

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/jdbc_test/JdbcThinLobTestApplication.java

@@ -0,0 +1,168 @@
+/*
+ * 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.ignite.internal.ducktest.tests.jdbc_test;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryPoolMXBean;
+import java.lang.management.MemoryUsage;
+import java.sql.Blob;
+import java.sql.Clob;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Optional;
+import java.util.Random;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+/**
+ * Simple application that used in smoke tests
+ */
+public class JdbcThinLobTestApplication extends IgniteAwareApplication {
+    /** {@inheritDoc} */
+    @Override public void run(JsonNode jsonNode) throws ClassNotFoundException, SQLException {
+        int blobSize = Optional.ofNullable(jsonNode.get("blob_size")).map(JsonNode::asInt).orElse(1);
+        int clobSize = Optional.ofNullable(jsonNode.get("clob_size")).map(JsonNode::asInt).orElse(1);
+        String action = Optional.ofNullable(jsonNode.get("action")).map(JsonNode::asText).orElse("insert");
+
+        int id = 1;
+
+        try (Connection conn = thinJdbcDataSource.getConnection()) {
+            conn.createStatement().execute("CREATE TABLE IF NOT EXISTS query(id INT, clob VARCHAR, blob BINARY, PRIMARY KEY(id)) " +
+                    "WITH \"cache_name=query,template=WITH_STATISTICS_ENABLED\"");
+
+            if ("insert".equals(action)) {
+                PreparedStatement insertStatement = conn.prepareStatement("INSERT INTO query(id, clob, blob) VALUES(?, ?, ?)");
+
+                insertStatement.setInt(1, id);
+
+                Clob clob = getClob(conn, clobSize);
+                insertStatement.setClob(2, clob);
+
+                Blob blob = getBlob(conn, blobSize);
+                insertStatement.setBlob(3, blob);
+
+                insertStatement.execute();
+            }
+            else if ("select".equals(action)) {
+                PreparedStatement selectStatement = conn.prepareStatement("SELECT * FROM query WHERE id = ?");
+
+                selectStatement.setInt(1, id);
+
+                ResultSet resultSet = selectStatement.executeQuery();
+
+                while (resultSet.next()) {
+                    Clob clob = resultSet.getClob("clob");
+
+                    recordResult("CLOB_SIZE", clob.length());
+                    recordResult("CLOB", clob.getSubString(1, Math.min((int)clob.length(), 64)));
+
+                    Blob blob = resultSet.getBlob("blob");
+                    byte[] bytes = blob.getBytes(1, Math.min((int)blob.length(), 64));
+
+                    recordResult("BLOB_SIZE", blob.length());
+                    recordResult("BLOB", U.byteArray2String(bytes, "0x%02X", ",0x%02X"));
+                }
+            }
+
+            markInitialized();
+
+            while (!terminated()) {
+                try {
+                    U.sleep(100); // Keeping node/txs alive.
+                }
+                catch (IgniteInterruptedCheckedException ignored) {
+                    log.info("Waiting interrupted.");
+                }
+            }
+
+            recordMemoryPeakUsage();
+
+            markFinished();
+        }
+    }
+
+    /**
+     * @param conn Connection.
+     * @param size CLOB size.
+     * @return CLOB of specified size.
+     */
+    private Clob getClob(Connection conn, int size) throws SQLException {
+        Clob clob = conn.createClob();
+
+        if (size > 0) {
+            StringBuilder sb = new StringBuilder(size);
+
+            for (int i = 0; i < size; i++) {
+                sb.append("Ж");
+            }
+
+            clob.setString(1, sb.toString());
+        }
+        return clob;
+    }
+
+    /**
+     * @param conn Connection.
+     * @param size BLOB size.
+     * @return BLOB with random data.
+     */
+    private Blob getBlob(Connection conn, int size) throws SQLException {
+        Blob blob = conn.createBlob();
+
+        if (size > 0) {
+            byte[] bytes = new byte[size];
+
+            new Random().nextBytes(bytes);
+
+            blob.setBytes(1, bytes);
+        }
+
+        return blob;
+    }
+
+    /**
+     * @return Memory usage.
+     */
+    private MemoryUsage getPeakMemoryUsage() {
+        List<MemoryPoolMXBean> pools = ManagementFactory.getMemoryPoolMXBeans();
+
+        return pools.stream().filter(pool -> pool.getName().contains("G1"))
+                .map(MemoryPoolMXBean::getPeakUsage)
+                .reduce((MemoryUsage a, MemoryUsage b) ->
+                        new MemoryUsage(a.getInit() + b.getInit(), a.getUsed() + b.getUsed(),
+                                a.getCommitted() + b.getCommitted(),
+                                Math.max(a.getMax() + b.getMax(), a.getCommitted() + b.getCommitted())))
+                .orElse(null);
+    }
+
+    /**
+     */
+    private void recordMemoryPeakUsage() {
+        MemoryUsage peakMemoryUsage = getPeakMemoryUsage();
+
+        recordResult("PEAK_MEMORY_USED", peakMemoryUsage.getUsed());
+        recordResult("PEAK_MEMORY_COMMITTED", peakMemoryUsage.getCommitted());
+        recordResult("PEAK_MEMORY_MAX", peakMemoryUsage.getMax());
+        recordResult("PEAK_MEMORY_INIT", peakMemoryUsage.getInit());
+    }
+}

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/utils/IgniteAwareApplication.java

@@ -21,6 +21,7 @@
 import java.lang.management.ThreadInfo;
 import com.fasterxml.jackson.databind.JsonNode;
 import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteJdbcThinDataSource;
 import org.apache.ignite.client.IgniteClient;
 import org.apache.ignite.cluster.ClusterState;
 import org.apache.ignite.internal.IgniteEx;
@@ -71,6 +72,9 @@ public abstract class IgniteAwareApplication {
     /** Client. */
     protected IgniteClient client;
 
+    /** Thin JDBC DataSource. */
+    protected IgniteJdbcThinDataSource thinJdbcDataSource;
+
     /** Cfg path. */
     protected String cfgPath;
 

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/utils/IgniteAwareApplicationService.java

@@ -47,7 +47,10 @@ private enum IgniteServiceType {
         THIN_CLIENT,
 
         /** Run application without precreated connections. */
-        NONE
+        NONE,
+
+        /** Run application with the precreated thin JDBC data source. */
+        THIN_JDBC
     }
 
     /**
@@ -106,6 +109,13 @@ else if (svcType == IgniteServiceType.THIN_CLIENT) {
         }
         else if (svcType == IgniteServiceType.NONE)
             app.start(jsonNode);
+        else if (svcType == IgniteServiceType.THIN_JDBC) {
+            log.info("Create thin jdbc ignite data source...");
+
+            app.thinJdbcDataSource = IgnitionEx.loadSpringBean(cfgPath, "thin.jdbc.cfg");
+
+            app.start(jsonNode);
+        }
         else
             throw new IllegalArgumentException("Unknown service type " + svcType);
     }

modules/ducktests/tests/ignitetest/services/utils/__init__.py

@@ -29,3 +29,4 @@ class IgniteServiceType(IntEnum):
     NODE = 0
     THIN_CLIENT = 1
     NONE = 2
+    THIN_JDBC = 3

modules/ducktests/tests/ignitetest/services/utils/config_template.py

@@ -24,6 +24,7 @@
 ZK_TEMPLATE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "zk", "templates")
 DEFAULT_IGNITE_CONF = "ignite.xml.j2"
 DEFAULT_THIN_CLIENT_CONF = "thin_client_config.xml.j2"
+DEFAULT_THIN_JDBC_CONF = "thin_jdbc_config.xml.j2"
 DEFAULT_LOG4J2_CONF = "log4j2.xml.j2"
 
 TEMPLATE_PATHES = [IGNITE_TEMPLATE_PATH, ZK_TEMPLATE_PATH]
@@ -85,6 +86,14 @@ def __init__(self, path=DEFAULT_THIN_CLIENT_CONF):
         super().__init__(path)
 
 
+class IgniteThinJdbcConfigTemplate(ConfigTemplate):
+    """
+    Ignite client node configuration.
+    """
+    def __init__(self, path=DEFAULT_THIN_JDBC_CONF):
+        super().__init__(path)
+
+
 class IgniteLoggerConfigTemplate(ConfigTemplate):
     """
     Ignite logger configuration.

modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py

@@ -101,7 +101,8 @@ def await_started(self, nodes=None):
         """
         Awaits start finished.
         """
-        if self.config.service_type in (IgniteServiceType.NONE, IgniteServiceType.THIN_CLIENT):
+        if self.config.service_type in (IgniteServiceType.NONE, IgniteServiceType.THIN_CLIENT,
+                                        IgniteServiceType.THIN_JDBC):
             return
 
         self.logger.info("Waiting for IgniteAware(s) to start ...")

modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/__init__.py

@@ -166,3 +166,35 @@ def service_type(self):
         Application mode.
         """
         return IgniteServiceType.THIN_CLIENT
+
+
+class IgniteThinJdbcConfiguration(NamedTuple):
+    addresses: list = []
+    version: IgniteVersion = DEV_BRANCH
+    ssl_params: SslParams = None
+    username: str = None
+    password: str = None
+
+    def prepare_ssl(self, test_globals, shared_root):
+        """
+        Updates ssl configuration from globals.
+        """
+        ssl_params = None
+        if self.ssl_params is None and is_ssl_enabled(test_globals):
+            ssl_params = get_ssl_params(test_globals, shared_root, IGNITE_CLIENT_ALIAS)
+        if ssl_params:
+            return self._replace(ssl_params=ssl_params)
+        return self
+
+    def prepare_for_env(self, cluster, node):
+        """
+        Updates configuration based on current environment.
+        """
+        return self
+
+    @property
+    def service_type(self):
+        """
+        Application mode.
+        """
+        return IgniteServiceType.THIN_JDBC

modules/ducktests/tests/ignitetest/services/utils/ignite_spec.py

@@ -29,7 +29,7 @@
 
 from ignitetest.services.utils import IgniteServiceType
 from ignitetest.services.utils.config_template import IgniteClientConfigTemplate, IgniteServerConfigTemplate, \
-    IgniteLoggerConfigTemplate, IgniteThinClientConfigTemplate
+    IgniteLoggerConfigTemplate, IgniteThinClientConfigTemplate, IgniteThinJdbcConfigTemplate
 from ignitetest.services.utils.jvm_utils import create_jvm_settings, merge_jvm_settings
 from ignitetest.services.utils.path import get_home_dir, IgnitePathAware
 from ignitetest.services.utils.ssl.ssl_params import is_ssl_enabled
@@ -142,6 +142,9 @@ def config_templates(self):
         if self.service.config.service_type == IgniteServiceType.THIN_CLIENT:
             config_templates.append((IgnitePathAware.IGNITE_THIN_CLIENT_CONFIG_NAME, IgniteThinClientConfigTemplate()))
 
+        if self.service.config.service_type == IgniteServiceType.THIN_JDBC:
+            config_templates.append((IgnitePathAware.IGNITE_THIN_CLIENT_CONFIG_NAME, IgniteThinJdbcConfigTemplate()))
+
         return config_templates
 
     def extend_config(self, config):

modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py

@@ -99,7 +99,7 @@ def mbean_attribute(self, mbean, attr):
         :return: Attribute value
         """
         cmd = "echo $'open %s\\n get -b %s %s \\n close' | %s | sed 's/%s = \\(.*\\);/\\1/'" \
-              % (self.pid, mbean, attr, self.jmx_util_cmd, attr)
+              % (self.pid, mbean.replace(' ', '\\ '), attr, self.jmx_util_cmd, attr)
 
         return iter(s.strip() for s in self.__run_cmd(cmd))
 

modules/ducktests/tests/ignitetest/services/utils/templates/thin_jdbc_config.xml.j2

@@ -0,0 +1,46 @@
+{#
+ 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.
+#}
+
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns:util="http://www.springframework.org/schema/util" xsi:schemaLocation="
+       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+       http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
+    <bean class="org.apache.ignite.IgniteJdbcThinDataSource" id="thin.jdbc.cfg">
+        <property name="addresses">
+            <list>
+                {% for address in config.addresses %}
+                    <value>{{ address }}</value>
+                {% endfor %}
+            </list>
+        </property>
+
+        {% if config.username %}
+            <property name="username" value="{{ config.username }}"/>
+            <property name="password" value="{{ config.password }}"/>
+        {% endif %}
+
+        {% if config.ssl_params %}
+            <property name="sslMode" value="require"/>
+            <property name="sslClientCertificateKeyStoreUrl" value="{{ config.ssl_params.key_store_path }}"/>
+            <property name="sslClientCertificateKeyStorePassword" value="{{ config.ssl_params.key_store_password }}"/>
+            <property name="sslTrustCertificateKeyStoreUrl" value="{{ config.ssl_params.trust_store_path }}"/>
+            <property name="sslTrustCertificateKeyStorePassword" value="{{ config.ssl_params.trust_store_password }}"/>
+        {% endif %}
+    </bean>
+</beans>