Research part corrupt ducktests

modules/core/src/main/java/org/apache/ignite/IgniteJdbcThinDataSource.java

@@ -59,7 +59,11 @@ public class IgniteJdbcThinDataSource implements DataSource, Serializable {
         if (!F.isEmpty(pwd))
             props.put("password", pwd);
 
-        return IgniteJdbcThinDriver.register().connect(getUrl(), props);
+        String url = getUrl();
+
+        System.err.println("TEST | JDBC connection URL: " + url);
+
+        return IgniteJdbcThinDriver.register().connect(url, props);
     }
 
     /** {@inheritDoc} */
@@ -194,6 +198,9 @@ public String getUrl() {
     public void setUrl(String url) throws SQLException {
         props = new ConnectionPropertiesImpl();
 
+        // TODO: for test purposes
+        assert !props.isPartitionAwareness();
+
         props.setUrl(url);
     }
 

modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionPropertiesImpl.java

@@ -21,6 +21,8 @@
 import java.sql.DriverPropertyInfo;
 import java.sql.SQLException;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import java.util.Properties;
 import java.util.StringTokenizer;
 import org.apache.ignite.IgniteCheckedException;
@@ -316,6 +318,10 @@ public class ConnectionPropertiesImpl implements ConnectionProperties, Serializa
 
             HostAndPortRange[] addrs = getAddresses();
 
+            List<HostAndPortRange> ranges = Arrays.asList(addrs);
+            Collections.shuffle(ranges);
+            addrs = ranges.toArray(new HostAndPortRange[ranges.size()]);
+
             for (int i = 0; i < addrs.length; i++) {
                 if (i > 0)
                     sbUrl.append(',');
@@ -599,11 +605,17 @@ public class ConnectionPropertiesImpl implements ConnectionProperties, Serializa
 
     /** {@inheritDoc} */
     @Override public boolean isPartitionAwareness() {
+        // TODO: for testing purposes.
+        assert !partitionAwareness.value();
+
         return partitionAwareness.value();
     }
 
     /** {@inheritDoc} */
     @Override public void setPartitionAwareness(boolean partitionAwareness) {
+        // TODO: for testing purposes.
+        assert !partitionAwareness;
+
         this.partitionAwareness.setValue(partitionAwareness);
     }
 

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/MapCacheStoreStrategy.java → modules/core/src/main/java/org/apache/ignite/internal/processors/cache/MapCacheStoreStrategy.java

@@ -110,6 +110,9 @@ public class MapCacheStoreStrategy implements TestCacheStoreStrategy {
 
     /** Serializable {@link #map} backed cache store factory */
     public static class MapStoreFactory implements Factory<CacheStore<Object, Object>> {
+        /** */
+        private static final long serialVersionUID = 0L;
+
         /** {@inheritDoc} */
         @Override public CacheStore<Object, Object> create() {
             return new MapCacheStore();

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mex/MexCntApplication.java

@@ -0,0 +1,60 @@
+/*
+ * 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.mex;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication;
+
+/** */
+public class MexCntApplication extends IgniteAwareApplication {
+    /** {@inheritDoc} */
+    @Override public void run(JsonNode jsonNode) throws Exception {
+        final String tableName = jsonNode.get("tableName").asText();
+        markInitialized();
+
+        recordResult("tableRowsCnt", printCount(tableName));
+
+        markFinished();
+    }
+
+    /** */
+    protected int printCount(String tableName) throws Exception {
+        try (Connection conn = thinJdbcDataSource.getConnection()) {
+            try {
+                ResultSet rs = conn.createStatement().executeQuery("SELECT COUNT(1) FROM " + tableName);
+
+                rs.next();
+
+                int cnt = rs.getInt(1);
+
+                log.info("TEST | Table '" + tableName + "' contains " + cnt + " records.");
+
+                return cnt;
+            }
+            catch (Exception t) {
+                log.error("Failed to get records count from table '" + tableName + "'. Error: " + t.getMessage(), t);
+
+                markBroken(t);
+
+                throw t;
+            }
+        }
+    }
+}

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mex/MexLoadApplication.java

@@ -0,0 +1,172 @@
+/*
+ * 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.mex;
+
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import com.fasterxml.jackson.databind.JsonNode;
+
+/** */
+public class MexLoadApplication extends MexCntApplication {
+    /** */
+    private static final int WAIT_START_SECS = 20;
+
+    /** {@inheritDoc} */
+    @Override public void run(JsonNode jsonNode) throws Exception {
+        final int preloadDurSec = jsonNode.get("preloadDurSec").asInt();
+        final int threads = jsonNode.get("threads").asInt();
+        final String tableName = jsonNode.get("tableName").asText();
+        final String cacheName = jsonNode.get("cacheName").asText();
+        final boolean transaction = jsonNode.get("transaction").asBoolean();
+
+        createTable(tableName, cacheName);
+
+        final ForkJoinPool executor = new ForkJoinPool(threads);
+        final CountDownLatch initLatch = new CountDownLatch(threads);
+        final AtomicLong counter = new AtomicLong();
+        final AtomicBoolean preloaded = new AtomicBoolean();
+
+        log.info("TEST | Load pool parallelism=" + executor.getParallelism() + ", transaction=" + transaction);
+
+        for (int i = 0; i < threads; ++i) {
+            executor.submit(() -> {
+                final Random rnd = new Random();
+                boolean init = false;
+
+                while (active()) {
+                    try (Connection conn = thinJdbcDataSource.getConnection()) {
+                        if (transaction) {
+                            conn.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
+                            conn.setAutoCommit(false);
+                        }
+
+                        PreparedStatement ps = conn.prepareStatement("INSERT INTO " + tableName + " values(?,?,?)");
+
+                        while (active()) {
+                            for (int t = 0; t < (transaction ? 3 + rnd.nextInt(8) : 1); ++t) {
+                                long id = counter.incrementAndGet();
+
+                                ps.setLong(1, id);
+
+                                byte[] data = new byte[rnd.nextInt(2048)];
+                                rnd.nextBytes(data);
+                                ps.setString(2, new String(data));
+
+                                ps.setBigDecimal(3, BigDecimal.valueOf(rnd.nextDouble()));
+
+                                int res = ps.executeUpdate();
+
+                                if (res != 1)
+                                    throw new IllegalStateException("Failed to insert a row. The result is not 1.");
+                            }
+
+                            if (transaction)
+                                conn.commit();
+
+                            if (!init) {
+                                init = true;
+
+                                initLatch.countDown();
+                            }
+                        }
+                    }
+                    catch (Throwable th) {
+                        if (!preloaded.get()) {
+                            log.error("TEST | Failed to preload. Marking as broken.", th);
+
+                            markBroken(th);
+
+                            synchronized (this) {
+                                notifyAll();
+                            }
+                        }
+                        else
+                            log.info("TEST | Failed to load. Recreating connection. Err: " + th.getMessage());
+                    }
+                }
+            });
+        }
+
+        if (!active())
+            return;
+
+        if (!initLatch.await(WAIT_START_SECS, TimeUnit.SECONDS)) {
+            markBroken(new IllegalStateException("Failed to start loading."));
+
+            return;
+        }
+
+        log.info("TEST | Started " + threads + " loading threads. Preloading for " + preloadDurSec + " seconds...");
+
+        synchronized (this) {
+            wait(preloadDurSec * 1000);
+        }
+
+        preloaded.set(true);
+
+        markInitialized();
+
+        while (active()) {
+            synchronized (this) {
+                wait(100);
+            }
+        }
+
+        //printCount(tableName);
+
+        markFinished();
+    }
+
+    /** */
+    private void createTable(String tableName, String cacheName) throws Exception {
+        try (Connection conn = thinJdbcDataSource.getConnection()) {
+            conn.createStatement().execute("CREATE TABLE " + tableName + "(" +
+                "id INT, strVal VARCHAR, decVal DECIMAL, PRIMARY KEY(id)" +
+                ") WITH \"cache_name=" + cacheName + "\""
+            );
+
+            try {
+                ResultSet rs = conn.prepareStatement("SELECT count(1) FROM " + tableName).executeQuery();
+
+                rs.next();
+
+                int cnt = rs.getInt(1);
+
+                if (cnt == 0)
+                    log.info("TEST | Created table '" + tableName + "' over cache '" + cacheName + "'.");
+                else
+                    throw new IllegalStateException("Unexpected empty table rows number: " + cnt);
+            }
+            catch (Exception t) {
+                log.error("Failed to create table '" + tableName + "'.", t);
+
+                markBroken(t);
+
+                throw t;
+            }
+        }
+    }
+}