Merge remote-tracking branch 'origin/dev-pool_loadbalance_fix' into dev-pool_loadbalance_fix

Author: chengyitian

test/com/xxdb/LoadBalanceTest.java

@@ -415,6 +415,9 @@ public void Test_getConnection_enableHighAvailability_true_enableLoadBalance_fal
             DBConnection connection = new DBConnection();
             connection.connect(HOST, PORT, "admin", "123456",null,true,null,false,false);
             list.add(connection);
+           // BasicInt re = (BasicInt)connection.run("getNodePort()");
+           // System.out.println("current node is："+re);
+           // System.out.println("stop current node");
         }
         DBConnection connection1 = new DBConnection();
         connection1.connect(HOST, PORT, "admin", "123456",false);
@@ -533,7 +536,7 @@ public void Test_getConnection_enableHighAvailability_true_enableLoadBalance_fal
     @Test
     public void Test_DBConnectionPool_enableHighAvailability_false_loadBalance_false() throws SQLException, ClassNotFoundException, IOException, InterruptedException {
         DBConnectionPool pool1 = new ExclusiveDBConnectionPool(HOST,PORT,"admin","123456",100,false,false);
-        Thread.sleep(10000);
+        Thread.sleep(1000);
         DBConnection connection1 = new DBConnection();
         connection1.connect(HOST, PORT, "admin", "123456",false);
         connection1.run("sleep(3000)");
@@ -546,7 +549,7 @@ public void Test_DBConnectionPool_enableHighAvailability_false_loadBalance_false
     @Test
     public void Test_DBConnectionPool_enableHighAvailability_true_loadBalance_false() throws SQLException, ClassNotFoundException, IOException, InterruptedException {
         DBConnectionPool pool1 = new ExclusiveDBConnectionPool(HOST,PORT,"admin","123456",100,false,true);
-        Thread.sleep(10000);
+        Thread.sleep(1000);
         DBConnection connection1 = new DBConnection();
         connection1.connect(HOST, PORT, "admin", "123456",false);
         connection1.run("sleep(3000)");
@@ -579,6 +582,45 @@ public void Test_DBConnectionPool_enableHighAvailability_true_loadBalance_false_
             }
         }
     }
+    //@Test//The current node is unavailable
+    public void Test_DBConnectionPool_enableHighAvailability_true_loadBalance_false_2() throws SQLException, ClassNotFoundException, IOException, InterruptedException {
+        DBConnection controller_conn = new DBConnection();
+        controller_conn.connect(controller_host, controller_port, "admin", "123456");
+        class MyThread extends Thread {
+            @Override
+            public void run() {
+                try {
+                    DBConnectionPool pool1 = new ExclusiveDBConnectionPool(HOST,PORT,"admin","123456",100,false,true,null,null, false, false, false);
+                    Thread.sleep(1000);
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+            }
+        }
+        class MyThread1 extends Thread {
+            @Override
+            public void run() {
+                    try {
+                        controller_conn.run("try{stopDataNode('"+HOST+":"+PORT+"')}catch(ex){}");
+                        Thread.sleep(1000);
+                    } catch (Exception e) {
+                        // 捕获异常并打印错误信息
+                        System.err.println("Error executing task: " + e.getMessage());
+                    }
+            }
+        }
+        MyThread thread = new MyThread();
+        MyThread1 thread1 = new MyThread1();
+        thread.start();
+        Thread.sleep(15);
+        System.err.println("thread1开始运行 ");
+        thread1.start();
+        thread.join();
+        thread1.join();
+        controller_conn.run("try{stopDataNode('"+HOST+":"+PORT+"')}catch(ex){}");
+    }
     @Test
     public void Test_DBConnectionPool_enableHighAvailability_true_loadBalance_true_highAvailabilitySites_null() throws SQLException, ClassNotFoundException, IOException {
         DBConnectionPool pool1 = new ExclusiveDBConnectionPool(HOST,PORT,"admin","123456",100,true,true,null,null, false, false, false);
@@ -601,7 +643,7 @@ public void Test_DBConnectionPool_enableHighAvailability_true_loadBalance_true()
         DBConnectionPool pool1 = new ExclusiveDBConnectionPool(HOST,PORT,"admin","123456",100,true,true,ipports,null, false, false, false);
         DBConnection connection1 = new DBConnection();
         connection1.connect(HOST, PORT, "admin", "123456",true);
-        connection1.run("sleep(1000)");
+        connection1.run("sleep(2000)");
         BasicTable re = (BasicTable) connection1.run("select port ,connectionNum  from rpc(getControllerAlias(),getClusterPerf) where mode= 0");
         for (int i = 0; i < re.rows(); i++) {
             System.out.println("port:" + re.getColumn(0).get(i) + " connectionNum:" + re.getColumn(1).get(i));
@@ -617,34 +659,41 @@ public void Test_DBConnectionPool_enableHighAvailability_true_loadBalance_true_1
         DBConnection controller_conn = new DBConnection();
         controller_conn.connect(controller_host, controller_port, "admin", "123456");
         controller_conn.run("try{stopDataNode('"+HOST+":"+PORT+"')}catch(ex){}");
+        controller_conn.run("sleep(1000)");
         DBConnectionPool pool1 = new ExclusiveDBConnectionPool(HOST,PORT,"admin","123456",100,true,true,ipports,null, false, false, false);
         controller_conn.run("try{startDataNode('"+HOST+":"+PORT+"')}catch(ex){}");
+        controller_conn.run("sleep(1000)");
         DBConnection connection1 = new DBConnection();
         connection1.connect(HOST, PORT, "admin", "123456",false);
         connection1.run("sleep(3000)");
-        BasicTable re = (BasicTable) connection1.run("select port ,connectionNum  from rpc(getControllerAlias(),getClusterPerf) where mode= 0");
+        BasicTable node1 = (BasicTable)connection1.run("select port ,connectionNum  from rpc(getControllerAlias(),getClusterPerf) where mode= 0 and port ="+ipports[1].split(":")[1]);
+        System.out.println(node1.getString());
+        Assert.assertEquals(true, Integer.valueOf(node1.getColumn(1).get(0).toString())>=50);
+
+        BasicTable node2 = (BasicTable)connection1.run("select port ,connectionNum  from rpc(getControllerAlias(),getClusterPerf) where mode= 0 and port ="+ipports[2].split(":")[1]);
+        System.out.println(node2.getString());
+        Assert.assertEquals(true, Integer.valueOf(node2.getColumn(1).get(0).toString())>=25);
+        pool1.shutdown();
+    }
+    @Test
+    public void Test_DBConnectionPool_enableHighAvailability_false_loadBalance_true() throws SQLException, ClassNotFoundException, IOException {
+        DBConnectionPool pool1 = new ExclusiveDBConnectionPool(HOST,PORT,"admin","123456",100,true,false,null,null, false, false, false);
+        DBConnection connection1 = new DBConnection();
+        connection1.connect(HOST, PORT, "admin", "123456",false);
+        connection1.run("sleep(3000)");
+        BasicTable re = (BasicTable) connection1.run("select port ,connectionNum  from rpc(getControllerAlias(),getClusterPerf) where mode in [0,4];");
         for (int i = 0; i < re.rows(); i++) {
             System.out.println("port:" + re.getColumn(0).get(i) + " connectionNum:" + re.getColumn(1).get(i));
             String port = re.getColumn(0).get(i).toString();
             String connectionNum = re.getColumn(1).get(i).toString();
             if(Integer.valueOf(port)!=PORT) {
-                assertEquals(true, Integer.valueOf(connectionNum) > 25);
+                assertEquals(true, Integer.valueOf(connectionNum) > 20);
                 assertEquals(true, Integer.valueOf(connectionNum) < 50);
             }
         }
         pool1.shutdown();
     }
     @Test
-    public void Test_DBConnectionPool_enableHighAvailability_false_loadBalance_true() throws SQLException, ClassNotFoundException, IOException {
-        String re = null;
-        try{
-            DBConnectionPool pool1 = new ExclusiveDBConnectionPool(HOST,PORT,"admin","123456",100,true,false,null,null, false, false, false);
-        }catch(Exception ex){
-            re = ex.getMessage();
-        }
-        Assert.assertEquals("Cannot only enable loadbalance but not enable highAvailablity.",re);
-    }
-    @Test
     public void Test_DBConnectionPool_enableHighAvailability_true_loadBalance_false_site_not_null() throws SQLException, ClassNotFoundException, IOException {
         DBConnection controller_conn = new DBConnection();
         controller_conn.connect(controller_host, controller_port, "admin", "123456");
@@ -663,16 +712,16 @@ public void Test_DBConnectionPool_enableHighAvailability_true_loadBalance_false_
         controller_conn.run("5000");
         List<DBTask> tasks = new ArrayList<>();
         for (int i = 0; i < 100; i++){
-            BasicDBTask task = new BasicDBTask("t = streamTable(10:0,`a`b,[INT,INT]);\n insert into t values(1,1);");
+            BasicDBTask task = new BasicDBTask("getNodePort();");
             tasks.add(task);
         }
         pool1.execute(tasks);
+        pool1.waitForThreadCompletion();
         BasicTable node2 = (BasicTable)controller_conn.run("select port ,connectionNum  from rpc(getControllerAlias(),getClusterPerf) where mode= 0 and port ="+PORT);
         System.out.println(node2.getString());
         Assert.assertEquals(true, Integer.valueOf(node2.getColumn(1).get(0).toString())>=100);
         controller_conn.run("try{startDataNode('"+HOST+":"+ipportArray[0].split(":")[1]+"')}catch(ex){}");
         controller_conn.run("2000");
         pool1.shutdown();
-
     }
 }

test/com/xxdb/Prepare.java

@@ -2,6 +2,7 @@
 
 import com.xxdb.data.BasicTable;
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.ResourceBundle;
 
 import static org.junit.Assert.assertEquals;
@@ -10,33 +11,36 @@ public class Prepare {
     static ResourceBundle bundle = ResourceBundle.getBundle("com/xxdb/setup/settings");
     static String HOST = bundle.getString("HOST");
     static int PORT = Integer.parseInt(bundle.getString("PORT"));
+    static int[] port_list = Arrays.stream(bundle.getString("PORTS").split(",")).mapToInt(Integer::parseInt).toArray();
 
     public static void clear_env() throws IOException {
-        DBConnection conn = new DBConnection();
-        conn.connect(HOST,PORT,"admin","123456");
-        conn.run("a = getStreamingStat().pubTables\n" +
-                "for(i in a){\n" +
-                "\ttry{stopPublishTable(i.subscriber.split(\":\")[0],int(i.subscriber.split(\":\")[1]),i.tableName,i.actions)}catch(ex){}\n" +
-                "}");
-        conn.run("def getAllShare(){\n" +
-                "\treturn select name from objs(true) where shared=1\n" +
-                "\t}\n" +
-                "\n" +
-                "def clearShare(){\n" +
-                "\tlogin(`admin,`123456)\n" +
-                "\tallShare=exec name from pnodeRun(getAllShare)\n" +
-                "\tfor(i in allShare){\n" +
-                "\t\ttry{\n" +
-                "\t\t\trpc((exec node from pnodeRun(getAllShare) where name =i)[0],clearTablePersistence,objByName(i))\n" +
-                "\t\t\t}catch(ex1){}\n" +
-                "\t\trpc((exec node from pnodeRun(getAllShare) where name =i)[0],undef,i,SHARED)\n" +
-                "\t}\n" +
-                "\ttry{\n" +
-                "\t\tPST_DIR=rpc(getControllerAlias(),getDataNodeConfig{getNodeAlias()})['persistenceDir']\n" +
-                "\t}catch(ex1){}\n" +
-                "}\n" +
-                "clearShare()");
-        conn.run("try{dropStreamEngine(\"serInput\");\n}catch(ex){\n}\n");
+        for (int i = 0; i < port_list.length; i++) {
+            DBConnection conn = new DBConnection();
+            conn.connect(HOST, port_list[i], "admin", "123456");
+            conn.run("a = getStreamingStat().pubTables\n" +
+                    "for(i in a){\n" +
+                    "\ttry{stopPublishTable(i.subscriber.split(\":\")[0],int(i.subscriber.split(\":\")[1]),i.tableName,i.actions)}catch(ex){}\n" +
+                    "}");
+            conn.run("def getAllShare(){\n" +
+                    "\treturn select name from objs(true) where shared=1\n" +
+                    "\t}\n" +
+                    "\n" +
+                    "def clearShare(){\n" +
+                    "\tlogin(`admin,`123456)\n" +
+                    "\tallShare=exec name from pnodeRun(getAllShare)\n" +
+                    "\tfor(i in allShare){\n" +
+                    "\t\ttry{\n" +
+                    "\t\t\trpc((exec node from pnodeRun(getAllShare) where name =i)[0],clearTablePersistence,objByName(i))\n" +
+                    "\t\t\t}catch(ex1){}\n" +
+                    "\t\trpc((exec node from pnodeRun(getAllShare) where name =i)[0],undef,i,SHARED)\n" +
+                    "\t}\n" +
+                    "\ttry{\n" +
+                    "\t\tPST_DIR=rpc(getControllerAlias(),getDataNodeConfig{getNodeAlias()})['persistenceDir']\n" +
+                    "\t}catch(ex1){}\n" +
+                    "}\n" +
+                    "clearShare()");
+            conn.run("try{dropStreamEngine(\"serInput\");\n}catch(ex){\n}\n");
+        }
     }
 
     public static void Preparedata(long count) throws IOException {

test/com/xxdb/streaming/client/cep/EventClientTest.java

@@ -856,9 +856,11 @@ public  void test_EventClient_subscribe_unsubscribe_resubscribe() throws IOExcep
         for(int i=0;i<10;i++) {
             client.subscribe(HOST, PORT, "inputTable", "test1", handler, -1, true, "admin", "123456");
             sender.sendEvent("MarketData", attributes);
+            Thread.sleep(200);
             client.unsubscribe(HOST, PORT, "inputTable", "test1");
             client.subscribe(HOST, PORT, "inputTable", "test1", handler, -1, true, "admin", "123456");
             sender.sendEvent("MarketData", attributes);
+            Thread.sleep(200);
             client.unsubscribe(HOST, PORT, "inputTable", "test1");
         }
         Thread.sleep(1000);
@@ -896,15 +898,27 @@ public  void test_EventClient_unsubscribe_duplicated() throws IOException, Inter
 
     @Test
     public  void test_EventClient_subscribe_haStreamTable() throws IOException, InterruptedException {
-        conn.run("table = table(1000000:0, `timestamp`eventType`event`comment1, [TIMESTAMP,STRING,BLOB,STRING]) as inputTable;");
-        conn.run("haStreamTable(11, table, `inputTable, 100000)");
-        conn.run("share table(100:0, `timestamp`comment1, [TIMESTAMP,STRING]) as outputTable;");
-        subscribePrepare();
+        String script = "try{\ndropStreamTable(`inputTable)\n}catch(ex){\n}\n"+
+                "table = table(1000000:0, `timestamp`eventType`event`comment1, [TIMESTAMP,STRING,BLOB,STRING]);\n"+
+                "haStreamTable("+GROUP_ID+", table, `inputTable, 100000);\n"+
+                "share table(100:0, `timestamp`comment1, [TIMESTAMP,STRING]) as outputTable;\n";
+        conn.run(script);
+        EventSchema scheme = new EventSchema();
+        scheme.setEventType("MarketData");
+        scheme.setFieldNames(Arrays.asList("timestamp", "comment1"));
+        scheme.setFieldTypes(Arrays.asList( DT_TIMESTAMP,DT_STRING));
+        scheme.setFieldForms(Arrays.asList(DF_SCALAR, DF_SCALAR));
+        List<EventSchema> eventSchemas = new ArrayList<>();
+        eventSchemas.add(scheme);
+        List<String> eventTimeKeys = Arrays.asList(new String[]{"timestamp"});
+        List<String> commonKeys = Arrays.asList(new String[]{"comment1"});
+        sender = new EventSender(conn, "inputTable", eventSchemas, eventTimeKeys, commonKeys);
+        client = new EventClient(eventSchemas, eventTimeKeys, commonKeys);
 
         List<Entity> attributes = new ArrayList<>();
         attributes.add(new BasicTimestamp(LocalDateTime.of(2024,3,22,10,45,3,100000000)));
         attributes.add(new BasicString("123456"));
-        client.subscribe(HOST, PORT, "inputTable", "test1", handler, -1, true, "user1", "123456");
+        client.subscribe(HOST, PORT, "inputTable", "test1", handler, -1, true, "admin", "123456");
         sender.sendEvent("MarketData", attributes);
         Thread.sleep(1000);
         BasicTable re = (BasicTable)conn.run("select * from outputTable");
@@ -927,12 +941,22 @@ public  void test_EventClient_subscribe_haStreamTable_leader() throws IOExceptio
             "haStreamTable("+GROUP_ID+", table, `inputTable_1, 100000);\n"+
             "share table(100:0, `timestamp`comment1, [TIMESTAMP,STRING]) as outputTable;\n";
         conn1.run(script);
-        subscribePrepare();
+        EventSchema scheme = new EventSchema();
+        scheme.setEventType("MarketData");
+        scheme.setFieldNames(Arrays.asList("timestamp", "comment1"));
+        scheme.setFieldTypes(Arrays.asList( DT_TIMESTAMP,DT_STRING));
+        scheme.setFieldForms(Arrays.asList(DF_SCALAR, DF_SCALAR));
+        List<EventSchema> eventSchemas = new ArrayList<>();
+        eventSchemas.add(scheme);
+        List<String> eventTimeKeys = Arrays.asList(new String[]{"timestamp"});
+        List<String> commonKeys = Arrays.asList(new String[]{"comment1"});
+        sender = new EventSender(conn, "inputTable_1", eventSchemas, eventTimeKeys, commonKeys);
+        client = new EventClient(eventSchemas, eventTimeKeys, commonKeys);
 
         List<Entity> attributes = new ArrayList<>();
         attributes.add(new BasicTimestamp(LocalDateTime.of(2024,3,22,10,45,3,100000000)));
         attributes.add(new BasicString("123456"));
-        client.subscribe(StreamLeaderHost, StreamLeaderPort, "inputTable_1", "test1", handler, -1, true, "user1", "123456");
+        client.subscribe(StreamLeaderHost, StreamLeaderPort, "inputTable_1", "test1", handler, -1, true, "admin", "123456");
         sender.sendEvent("MarketData", attributes);
         Thread.sleep(1000);
         BasicTable re = (BasicTable)conn1.run("select * from outputTable");
@@ -942,7 +966,7 @@ public  void test_EventClient_subscribe_haStreamTable_leader() throws IOExceptio
         client.unsubscribe(StreamLeaderHost, StreamLeaderPort, "inputTable_1", "test1");
     }
 
-    @Test//not support
+    //@Test//not support
     public  void test_EventClient_subscribe_haStreamTable_follower() throws IOException, InterruptedException {
         String script0 ="leader = getStreamingLeader("+GROUP_ID+");\n" +
                 "groupSitesStr = (exec sites from getStreamingRaftGroups() where id =="+GROUP_ID+")[0];\n"+
@@ -962,8 +986,18 @@ public  void test_EventClient_subscribe_haStreamTable_follower() throws IOExcept
                 "haStreamTable("+GROUP_ID+", table, `inputTable_1, 100000);\n"+
                 "share table(100:0, `timestamp`comment1, [TIMESTAMP,STRING]) as outputTable;\n";
         conn1.run(script);
+        EventSchema scheme = new EventSchema();
+        scheme.setEventType("MarketData");
+        scheme.setFieldNames(Arrays.asList("timestamp", "comment1"));
+        scheme.setFieldTypes(Arrays.asList( DT_TIMESTAMP,DT_STRING));
+        scheme.setFieldForms(Arrays.asList(DF_SCALAR, DF_SCALAR));
+        List<EventSchema> eventSchemas = new ArrayList<>();
+        eventSchemas.add(scheme);
+        List<String> eventTimeKeys = Arrays.asList(new String[]{"timestamp"});
+        List<String> commonKeys = Arrays.asList(new String[]{"comment1"});
+        sender = new EventSender(conn, "inputTable_1", eventSchemas, eventTimeKeys, commonKeys);
+        client = new EventClient(eventSchemas, eventTimeKeys, commonKeys);
 
-        subscribePrepare();
         List<Entity> attributes = new ArrayList<>();
         attributes.add(new BasicTimestamp(LocalDateTime.of(2024,3,22,10,45,3,100000000)));
         attributes.add(new BasicString("123456"));