From 0ff1558cbc8a13e86b72e7497bb8b6b071609de7 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Thu, 29 Oct 2026 21:22:13 +0000 Subject: [PATCH 1/2] [tests] Wait until users, roles and grants reach all cluster replicas On the 3-replica test cluster, users, roles, settings profiles and grants are kept in the replicated user directory. Each replica updates its own copy from Keeper asynchronously, and nginx sends each request to any replica. A login, GRANT or DROP ... IF EXISTS right after a change can go to a replica that does not have the change yet: AUTHENTICATION_FAILED, UNKNOWN_ROLE, or ACCESS_ENTITY_ALREADY_EXISTS. Add BaseIntegrationTest.waitForAllReplicas(). It checks system. on every replica through clusterAllReplicas('test_cluster', ...) with a limited number of attempts, and does nothing outside cluster mode. Call it after each access change that a later request depends on, and give each test its own user, role and profile names, dropped in finally. Part of: https://github.com/ClickHouse/clickhouse-java/issues/3267 Fixes: https://github.com/ClickHouse/clickhouse-java/issues/3237 Co-Authored-By: Claude Opus 5.5 --- .../client/BaseIntegrationTest.java | 52 +++++ .../client/ClickHouseServerForTest.java | 45 ++++ .../com/clickhouse/client/ClientTests.java | 102 +++++---- .../clickhouse/client/HttpTransportTests.java | 130 +++++------ .../clickhouse/jdbc/ReadonlyProfileTest.java | 39 ++-- .../com/clickhouse/jdbc/StatementTest.java | 202 ++++++++++-------- .../ShowStatementDatabaseMetaDataTest.java | 29 +-- 7 files changed, 365 insertions(+), 234 deletions(-) diff --git a/clickhouse-client/src/test/java/com/clickhouse/client/BaseIntegrationTest.java b/clickhouse-client/src/test/java/com/clickhouse/client/BaseIntegrationTest.java index 046ab46ac..c1a0d72a0 100644 --- a/clickhouse-client/src/test/java/com/clickhouse/client/BaseIntegrationTest.java +++ b/clickhouse-client/src/test/java/com/clickhouse/client/BaseIntegrationTest.java @@ -1,9 +1,11 @@ package com.clickhouse.client; import com.clickhouse.client.config.ClickHouseClientOption; +import org.testng.Assert; import org.testng.annotations.AfterTest; import org.testng.annotations.BeforeTest; +import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.Locale; @@ -13,6 +15,9 @@ * Base class for all integration tests. */ public abstract class BaseIntegrationTest { + private static final int REPLICA_CHECK_ATTEMPTS = 100; + private static final long REPLICA_CHECK_INTERVAL_MS = 100L; + @BeforeTest(groups = {"integration"}) public static void setupClickHouseContainer() { ClickHouseServerForTest.beforeSuite(); @@ -86,6 +91,53 @@ protected static String onCluster() { return ClickHouseServerForTest.isCluster() ? " ON CLUSTER 'test_cluster'" : ""; } + /** + * Waits until each replica of the test cluster returns {@code expectedRows} rows of + * {@code system.
} that match {@code condition}, so that the next request sees a + * created or changed user, role, settings profile or grant on any replica. + * Does nothing when the tests do not run on a cluster. + */ + protected static void waitForAllReplicas(String table, String condition, int expectedRows) { + if (!isCluster()) { + return; + } + String sql = "SELECT hostName() AS host, countIf(" + condition + ") AS matching_rows FROM clusterAllReplicas('" + + ClusterClickHouseContainerStarter.CLUSTER_NAME + "', system." + table + ") GROUP BY host"; + String state = ""; + for (int attempt = 1; attempt <= REPLICA_CHECK_ATTEMPTS; attempt++) { + try { + state = ClickHouseServerForTest.queryForResult(sql).trim(); + if (hasRowsOnAllReplicas(state, expectedRows)) { + return; + } + } catch (IOException e) { + state = e.getMessage(); + } + try { + Thread.sleep(REPLICA_CHECK_INTERVAL_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + Assert.fail("Expected " + expectedRows + " rows matching [" + condition + "] in system." + table + " on all " + + ClusterClickHouseContainerStarter.NODE_COUNT + " replicas, but got [host, rows]:\n" + state); + } + + private static boolean hasRowsOnAllReplicas(String result, int expectedRows) { + String[] lines = result.split("\n"); + if (lines.length != ClusterClickHouseContainerStarter.NODE_COUNT) { + return false; + } + for (String line : lines) { + String[] columns = line.split("\t"); + if (columns.length != 2 || !columns[1].equals(String.valueOf(expectedRows))) { + return false; + } + } + return true; + } + protected boolean createDatabase(String dbName) { return ClickHouseServerForTest.runQuery("CREATE DATABASE IF NOT EXISTS `" + dbName + "`" + onCluster()); } diff --git a/clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseServerForTest.java b/clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseServerForTest.java index bd8bd0871..4e452d087 100644 --- a/clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseServerForTest.java +++ b/clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseServerForTest.java @@ -9,6 +9,9 @@ import com.clickhouse.client.ClickHouseTestEnvironment.ContainerMode; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; import java.io.OutputStream; import java.net.HttpURLConnection; import java.net.InetSocketAddress; @@ -279,6 +282,48 @@ public static boolean runQuery(String sql) { return false; } + /** + * Runs a query once and returns the response body (TabSeparated). + * + * @throws IOException when the request fails or the server returns an error + */ + public static String queryForResult(String sql) throws IOException { + ClickHouseNode server = getClickHouseNode(ClickHouseProtocol.HTTP, isCloud(), ClickHouseNode.builder().build()); + byte[] postData = sql.getBytes(StandardCharsets.UTF_8); + String authorization = Base64.getEncoder() + .encodeToString((getUsername() + ":" + getPassword()).getBytes(StandardCharsets.UTF_8)); + HttpURLConnection httpConn = (HttpURLConnection) new URL(server.getBaseUri()).openConnection(); + try { + httpConn.setConnectTimeout(10000); + httpConn.setReadTimeout(10000); + httpConn.setRequestMethod("POST"); + httpConn.setDoOutput(true); + httpConn.setRequestProperty("Authorization", "Basic " + authorization); + httpConn.setFixedLengthStreamingMode(postData.length); + try (OutputStream out = httpConn.getOutputStream()) { + out.write(postData, 0, postData.length); + } + + boolean ok = httpConn.getResponseCode() == HttpURLConnection.HTTP_OK; + ByteArrayOutputStream body = new ByteArrayOutputStream(); + try (InputStream in = ok ? httpConn.getInputStream() : httpConn.getErrorStream()) { + if (in != null) { + byte[] buffer = new byte[4096]; + for (int read; (read = in.read(buffer)) != -1; ) { + body.write(buffer, 0, read); + } + } + } + String result = new String(body.toByteArray(), StandardCharsets.UTF_8); + if (!ok) { + throw new IOException("HTTP " + httpConn.getResponseCode() + ": " + result); + } + return result; + } finally { + httpConn.disconnect(); + } + } + private static Endpoint resolve(ClickHouseProtocol protocol, int port, boolean applyOverride) { if (starter != null) { int containerPort = (port == ClickHouseProtocol.HTTP.getDefaultPort()) diff --git a/client-v2/src/test/java/com/clickhouse/client/ClientTests.java b/client-v2/src/test/java/com/clickhouse/client/ClientTests.java index 2d8f8ad85..e0c7c5d0c 100644 --- a/client-v2/src/test/java/com/clickhouse/client/ClientTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/ClientTests.java @@ -434,28 +434,43 @@ private static Object[][] sessionRoles() { } @Test(groups = {"integration"}, dataProvider = "sessionRoles") - public void testOperationCustomRoles(String[] roles) throws Exception { + public void testOperationCustomRoles(String[] roleNames) throws Exception { if (isVersionMatch("(,24.3]", newClient().build())) { return; } + String suffix = "_" + RandomStringUtils.insecure().next(8, true, true); + String user = "some_user" + suffix; + String[] roles = Arrays.stream(roleNames).map(role -> role + suffix).toArray(String[]::new); String password = "^1A" + RandomStringUtils.random(12, true, true) + "3b$"; final String rolesList = "\"" + Strings.join("\",\"", roles) + "\""; - try (Client client = newClient().build()) { - client.execute("DROP ROLE IF EXISTS " + rolesList).get().close(); - client.execute("CREATE ROLE " + rolesList).get().close(); - client.execute("DROP USER IF EXISTS some_user").get().close(); - client.execute("CREATE USER some_user IDENTIFIED BY '" + password + "'").get().close(); - client.execute("GRANT " + rolesList + " TO some_user").get().close(); + try (Client client = newClient().build(); + Client userClient = newClient().setUsername(user).setPassword(password).build()) { + try { + createUserWithRoles(client, user, "IDENTIFIED BY '" + password + "'", roles); + QuerySettings settings = new QuerySettings().setDBRoles(Arrays.asList(roles)); + List resp = userClient.queryAll("SELECT currentRoles()", settings); + Set roleSet = new HashSet<>(Arrays.asList(roles)); + Set currentRoles = new HashSet (resp.get(0).getList(1)); + Assert.assertEquals(currentRoles, roleSet, "Roles " + roleSet + " not found in " + currentRoles); + } finally { + client.execute("DROP USER IF EXISTS " + user).get().close(); + client.execute("DROP ROLE IF EXISTS " + rolesList).get().close(); + } } + } - try (Client userClient = newClient().setUsername("some_user").setPassword(password).build()) { - QuerySettings settings = new QuerySettings().setDBRoles(Arrays.asList(roles)); - List resp = userClient.queryAll("SELECT currentRoles()", settings); - Set roleSet = new HashSet<>(Arrays.asList(roles)); - Set currentRoles = new HashSet (resp.get(0).getList(1)); - Assert.assertEquals(currentRoles, roleSet, "Roles " + roleSet + " not found in " + currentRoles); - } + private static void createUserWithRoles(Client client, String user, String identification, String[] roles) + throws Exception { + final String rolesList = "\"" + Strings.join("\",\"", roles) + "\""; + final String roleNames = "'" + Strings.join("','", roles) + "'"; + client.execute("CREATE ROLE " + rolesList).get().close(); + client.execute("CREATE USER " + user + " " + identification).get().close(); + waitForAllReplicas("roles", "name IN (" + roleNames + ")", roles.length); + waitForAllReplicas("users", "name = '" + user + "'", 1); + client.execute("GRANT " + rolesList + " TO " + user).get().close(); + waitForAllReplicas("role_grants", "user_name = '" + user + "' AND granted_role_name IN (" + roleNames + ")", + roles.length); } @DataProvider(name = "clientSessionRoles") @@ -466,27 +481,29 @@ private static Object[][] clientSessionRoles() { }; } @Test(groups = {"integration"}, dataProvider = "clientSessionRoles") - public void testClientCustomRoles(String[] roles) throws Exception { + public void testClientCustomRoles(String[] roleNames) throws Exception { if (isVersionMatch("(,24.3]", newClient().build())) { return; } + String suffix = "_" + RandomStringUtils.insecure().next(8, true, true); + String user = "some_user" + suffix; + String[] roles = Arrays.stream(roleNames).map(role -> role + suffix).toArray(String[]::new); String password = "^1A" + RandomStringUtils.insecure().next(12, true, true) + "3B$"; final String rolesList = "\"" + Strings.join("\",\"", roles) + "\""; - try (Client client = newClient().build()) { - client.execute("DROP ROLE IF EXISTS " + rolesList).get().close(); - client.execute("CREATE ROLE " + rolesList).get().close(); - client.execute("DROP USER IF EXISTS some_user").get().close(); - client.execute("CREATE USER some_user IDENTIFIED WITH sha256_password BY '" + password + "'").get().close(); - client.execute("GRANT " + rolesList + " TO some_user").get().close(); - } - - try (Client userClient = newClient().setUsername("some_user").setPassword(password).build()) { - userClient.setDBRoles(Arrays.asList(roles)); - List resp = userClient.queryAll("SELECT currentRoles()"); - Set roleSet = new HashSet<>(Arrays.asList(roles)); - Set currentRoles = new HashSet (resp.get(0).getList(1)); - Assert.assertEquals(currentRoles, roleSet, "Roles " + roleSet + " not found in " + currentRoles); + try (Client client = newClient().build(); + Client userClient = newClient().setUsername(user).setPassword(password).build()) { + try { + createUserWithRoles(client, user, "IDENTIFIED WITH sha256_password BY '" + password + "'", roles); + userClient.setDBRoles(Arrays.asList(roles)); + List resp = userClient.queryAll("SELECT currentRoles()"); + Set roleSet = new HashSet<>(Arrays.asList(roles)); + Set currentRoles = new HashSet (resp.get(0).getList(1)); + Assert.assertEquals(currentRoles, roleSet, "Roles " + roleSet + " not found in " + currentRoles); + } finally { + client.execute("DROP USER IF EXISTS " + user).get().close(); + client.execute("DROP ROLE IF EXISTS " + rolesList).get().close(); + } } } @@ -507,36 +524,15 @@ public void testRuntimeCredentialChange() throws Exception { adminClient.execute("DROP USER IF EXISTS " + user2).get().close(); adminClient.execute("CREATE USER " + user1 + " IDENTIFIED BY '" + password1 + "'").get().close(); adminClient.execute("CREATE USER " + user2 + " IDENTIFIED BY '" + password2 + "'").get().close(); + waitForAllReplicas("users", "name IN ('" + user1 + "', '" + user2 + "')", 2); try (Client userClient = newClient().setUsername(user1).setPassword(password1).build()) { - List firstResponse = null; - for (int i = 0; i < 20; i++) { - try { - firstResponse = userClient.queryAll("SELECT currentUser() AS user"); - break; - } catch (ClientException e) { - if (i == 19) { - throw e; - } - Thread.sleep(250); - } - } + List firstResponse = userClient.queryAll("SELECT currentUser() AS user"); Assert.assertEquals(firstResponse.get(0).getString("user"), user1); userClient.updateUserAndPassword(user2, password2); - List secondResponse = null; - for (int i = 0; i < 20; i++) { - try { - secondResponse = userClient.queryAll("SELECT currentUser() AS user"); - break; - } catch (ClientException e) { - if (i == 19) { - throw e; - } - Thread.sleep(250); - } - } + List secondResponse = userClient.queryAll("SELECT currentUser() AS user"); Assert.assertEquals(secondResponse.get(0).getString("user"), user2); } } finally { diff --git a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java index 533a43268..8a32fcd9b 100644 --- a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java @@ -1044,29 +1044,31 @@ public void testSSLAuthentication() throws Exception { return; // Current test is working only with local server because of self-signed certificates. } ClickHouseNode server = getSecureServer(ClickHouseProtocol.HTTP); - try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), true) + String user = "some_user_" + UUID.randomUUID().toString().replace("-", ""); + try (Client adminClient = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), true) .setUsername("dba") .setPassword("dba") .setRootCertificate("containers/clickhouse-server/certs/localhost.crt") .build()) { - try (CommandResponse resp = client.execute("DROP USER IF EXISTS some_user").get()) { - } - try (CommandResponse resp = client.execute("CREATE USER some_user IDENTIFIED WITH ssl_certificate CN 'some_user'").get()) { - } - } + try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), true) + .useSSLAuthentication(true) + .setUsername(user) + .setRootCertificate("containers/clickhouse-server/certs/localhost.crt") + .setClientCertificate("some_user.crt") + .setClientKey("some_user.key") + .compressServerResponse(false) + .build()) { - try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), true) - .useSSLAuthentication(true) - .setUsername("some_user") - .setRootCertificate("containers/clickhouse-server/certs/localhost.crt") - .setClientCertificate("some_user.crt") - .setClientKey("some_user.key") - .compressServerResponse(false) - .build()) { + try (CommandResponse resp = adminClient.execute("CREATE USER " + user + " IDENTIFIED WITH ssl_certificate CN 'some_user'").get()) { + } + waitForAllReplicas("users", "name = '" + user + "'", 1); - try (QueryResponse resp = client.query("SELECT 1").get()) { - Assert.assertEquals(resp.getReadRows(), 1); + try (QueryResponse resp = client.query("SELECT 1").get()) { + Assert.assertEquals(resp.getReadRows(), 1); + } + } finally { + adminClient.execute("DROP USER IF EXISTS " + user).get().close(); } } } @@ -1077,43 +1079,46 @@ public void testPasswordAuthentication(String identifyWith, String identifyBy, b return; // Current test is working only with local server because of self-signed certificates. } ClickHouseNode server = getServer(ClickHouseProtocol.HTTP); + String user = "some_user_" + UUID.randomUUID().toString().replace("-", ""); - try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) + try (Client adminClient = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) .setUsername("dba") .setPassword("dba") .build()) { - try (CommandResponse resp = client.execute("DROP USER IF EXISTS some_user").get()) { - } - try (CommandResponse resp = client.execute("CREATE USER some_user IDENTIFIED WITH " + identifyWith + " BY '" + identifyBy + "'").get()) { - } - } catch (Exception e) { - Assert.fail("Failed on setup", e); - } - + try { + try (CommandResponse resp = adminClient.execute("CREATE USER " + user + " IDENTIFIED WITH " + identifyWith + " BY '" + identifyBy + "'").get()) { + } catch (Exception e) { + Assert.fail("Failed on setup", e); + } + waitForAllReplicas("users", "name = '" + user + "'", 1); - try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) - .setUsername("some_user") - .setPassword(identifyBy) - .build()) { + try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) + .setUsername(user) + .setPassword(identifyBy) + .build()) { - Assert.assertEquals(client.queryAll("SELECT user()").get(0).getString(1), "some_user"); - } catch (Exception e) { - Assert.fail("Failed to authenticate", e); - } + Assert.assertEquals(client.queryAll("SELECT user()").get(0).getString(1), user); + } catch (Exception e) { + Assert.fail("Failed to authenticate", e); + } - if (failsWithHeaders) { - try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) - .setUsername("some_user") - .setPassword(identifyBy) - .useHTTPBasicAuth(false) - .build()) { + if (failsWithHeaders) { + try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) + .setUsername(user) + .setPassword(identifyBy) + .useHTTPBasicAuth(false) + .build()) { - Assert.expectThrows(ClientException.class, () -> - client.queryAll("SELECT user()").get(0).getString(1)); + Assert.expectThrows(ClientException.class, () -> + client.queryAll("SELECT user()").get(0).getString(1)); - } catch (Exception e) { - Assert.fail("Unexpected exception", e); + } catch (Exception e) { + Assert.fail("Unexpected exception", e); + } + } + } finally { + adminClient.execute("DROP USER IF EXISTS " + user).get().close(); } } } @@ -1141,30 +1146,33 @@ public void testAuthHeaderIsKeptFromUser() throws Exception { String identifyWith = "sha256_password"; String identifyBy = "123ยง"; - try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) + String user = "some_user_" + UUID.randomUUID().toString().replace("-", ""); + try (Client adminClient = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) .setUsername("dba") .setPassword("dba") .build()) { - try (CommandResponse resp = client.execute("DROP USER IF EXISTS some_user").get()) { - } - try (CommandResponse resp = client.execute("CREATE USER some_user IDENTIFIED WITH " + identifyWith + " BY '" + identifyBy + "'").get()) { - } - } catch (Exception e) { - Assert.fail("Failed on setup", e); - } - + try { + try (CommandResponse resp = adminClient.execute("CREATE USER " + user + " IDENTIFIED WITH " + identifyWith + " BY '" + identifyBy + "'").get()) { + } catch (Exception e) { + Assert.fail("Failed on setup", e); + } + waitForAllReplicas("users", "name = '" + user + "'", 1); - try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) - .setUsername("some_user") - .setPassword(identifyBy) - .useHTTPBasicAuth(false) // disable basic auth to produce CH headers - .httpHeader(HttpHeaders.AUTHORIZATION, "Basic " + Base64.getEncoder().encodeToString(("some_user:" +identifyBy).getBytes())) - .build()) { + try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) + .setUsername(user) + .setPassword(identifyBy) + .useHTTPBasicAuth(false) // disable basic auth to produce CH headers + .httpHeader(HttpHeaders.AUTHORIZATION, "Basic " + Base64.getEncoder().encodeToString((user + ":" +identifyBy).getBytes())) + .build()) { - Assert.assertEquals(client.queryAll("SELECT user()").get(0).getString(1), "some_user"); - } catch (Exception e) { - Assert.fail("Failed to authenticate", e); + Assert.assertEquals(client.queryAll("SELECT user()").get(0).getString(1), user); + } catch (Exception e) { + Assert.fail("Failed to authenticate", e); + } + } finally { + adminClient.execute("DROP USER IF EXISTS " + user).get().close(); + } } } diff --git a/jdbc-v2/src/test/java/com/clickhouse/jdbc/ReadonlyProfileTest.java b/jdbc-v2/src/test/java/com/clickhouse/jdbc/ReadonlyProfileTest.java index 2c34ed447..0c3322555 100644 --- a/jdbc-v2/src/test/java/com/clickhouse/jdbc/ReadonlyProfileTest.java +++ b/jdbc-v2/src/test/java/com/clickhouse/jdbc/ReadonlyProfileTest.java @@ -18,6 +18,10 @@ public class ReadonlyProfileTest extends JdbcIntegrationTest { private String password; + private String user1; + private String user2; + private String profile1; + private String profile2; @BeforeClass(groups = { "integration" }) public void setup() throws SQLException { @@ -27,18 +31,25 @@ public void setup() throws SQLException { // Append fixed character classes so the random password satisfies server // password-complexity policies (uppercase, lowercase, digit, special). password = UUID.randomUUID().toString() + "Aa1!"; + String suffix = "_" + UUID.randomUUID().toString().replace("-", ""); + user1 = "jdbc_test_user_readonly_1" + suffix; + user2 = "jdbc_test_user_readonly_2" + suffix; + profile1 = "jdbc_test_profile_readonly_1" + suffix; + profile2 = "jdbc_test_profile_readonly_2" + suffix; com.clickhouse.client.ClickHouseServerForTest.beforeSuite(); try (Connection conn = getJdbcConnection(); Statement stmt = conn.createStatement()) { - stmt.execute("CREATE SETTINGS PROFILE IF NOT EXISTS jdbc_test_profile_readonly_1 SETTINGS readonly=1"); - stmt.execute("CREATE SETTINGS PROFILE IF NOT EXISTS jdbc_test_profile_readonly_2 SETTINGS readonly=2"); - stmt.execute("DROP USER IF EXISTS jdbc_test_user_readonly_1"); - stmt.execute("DROP USER IF EXISTS jdbc_test_user_readonly_2"); - stmt.execute("CREATE USER jdbc_test_user_readonly_1 IDENTIFIED WITH plaintext_password BY '" + password + "' SETTINGS PROFILE jdbc_test_profile_readonly_1"); - stmt.execute("CREATE USER jdbc_test_user_readonly_2 IDENTIFIED WITH plaintext_password BY '" + password + "' SETTINGS PROFILE jdbc_test_profile_readonly_2"); + stmt.execute("CREATE SETTINGS PROFILE " + profile1 + " SETTINGS readonly=1"); + stmt.execute("CREATE SETTINGS PROFILE " + profile2 + " SETTINGS readonly=2"); + waitForAllReplicas("settings_profiles", "name IN ('" + profile1 + "', '" + profile2 + "')", 2); + stmt.execute("CREATE USER " + user1 + " IDENTIFIED WITH plaintext_password BY '" + password + "' SETTINGS PROFILE " + profile1); + stmt.execute("CREATE USER " + user2 + " IDENTIFIED WITH plaintext_password BY '" + password + "' SETTINGS PROFILE " + profile2); + String users = "'" + user1 + "', '" + user2 + "'"; + waitForAllReplicas("users", "name IN (" + users + ")", 2); String db = getDatabase(); - stmt.execute("GRANT SELECT ON " + db + ".* TO jdbc_test_user_readonly_1"); - stmt.execute("GRANT SELECT ON " + db + ".* TO jdbc_test_user_readonly_2"); + stmt.execute("GRANT SELECT ON " + db + ".* TO " + user1); + stmt.execute("GRANT SELECT ON " + db + ".* TO " + user2); + waitForAllReplicas("grants", "user_name IN (" + users + ")", 2); } } @@ -49,10 +60,10 @@ public void teardown() throws SQLException { } try (Connection conn = getJdbcConnection(); Statement stmt = conn.createStatement()) { - stmt.execute("DROP USER IF EXISTS jdbc_test_user_readonly_1"); - stmt.execute("DROP USER IF EXISTS jdbc_test_user_readonly_2"); - stmt.execute("DROP SETTINGS PROFILE IF EXISTS jdbc_test_profile_readonly_1"); - stmt.execute("DROP SETTINGS PROFILE IF EXISTS jdbc_test_profile_readonly_2"); + stmt.execute("DROP USER IF EXISTS " + user1); + stmt.execute("DROP USER IF EXISTS " + user2); + stmt.execute("DROP SETTINGS PROFILE IF EXISTS " + profile1); + stmt.execute("DROP SETTINGS PROFILE IF EXISTS " + profile2); } } @@ -62,7 +73,7 @@ public void testReadonly1CannotChangeSettings() throws Exception { throw new SkipException("Should not be tested in cloud because creates users and profiles"); } Properties properties = new Properties(); - properties.setProperty("user", "jdbc_test_user_readonly_1"); + properties.setProperty("user", user1); properties.setProperty("password", password); properties.put(ClientConfigProperties.serverSetting(ServerSettings.OUTPUT_FORMAT_BINARY_WRITE_JSON_AS_STRING), "1"); @@ -86,7 +97,7 @@ public void testReadonly2CanChangeSettings() throws Exception { } Properties properties = new Properties(); - properties.setProperty("user", "jdbc_test_user_readonly_2"); + properties.setProperty("user", user2); properties.setProperty("password", password); properties.put(ClientConfigProperties.serverSetting(ServerSettings.OUTPUT_FORMAT_BINARY_WRITE_JSON_AS_STRING), "1"); properties.put(ClientConfigProperties.serverSetting("allow_experimental_json_type"), "1"); diff --git a/jdbc-v2/src/test/java/com/clickhouse/jdbc/StatementTest.java b/jdbc-v2/src/test/java/com/clickhouse/jdbc/StatementTest.java index 0657558df..d753f466a 100644 --- a/jdbc-v2/src/test/java/com/clickhouse/jdbc/StatementTest.java +++ b/jdbc-v2/src/test/java/com/clickhouse/jdbc/StatementTest.java @@ -556,118 +556,130 @@ public void testSettingRole(SqlParserFacade.SQLParser parser) throws SQLExceptio return; } - List roles = Arrays.asList("role1", "role2", "role3"); + String suffix = "_" + RandomStringUtils.insecure().next(8, true, true); + String user = "some_user" + suffix; + String role1 = "role1" + suffix; + String role2 = "role2" + suffix; + String role3 = "role3" + suffix; + List roles = Arrays.asList(role1, role2, role3); + String roleNames = "'" + String.join("','", roles) + "'"; final String userPass = "^1A" + RandomStringUtils.random(12, true, true) + "3B$"; Properties properties = new Properties(); properties.setProperty(DriverProperties.SQL_PARSER.getKey(), parser.name()); - try (ConnectionImpl conn = (ConnectionImpl) getJdbcConnection(properties)) { - try (Statement stmt = conn.createStatement()) { - stmt.execute("DROP ROLE IF EXISTS " + String.join(", ", roles)); - stmt.execute("DROP USER IF EXISTS some_user"); - stmt.execute("CREATE ROLE " + String.join(", ", roles)); - stmt.execute("CREATE USER some_user IDENTIFIED BY '" + userPass + "'"); - stmt.execute("GRANT " + String.join(", ", roles) + " TO some_user"); - stmt.execute("SET DEFAULT ROLE NONE TO some_user"); - } - } + try (ConnectionImpl adminConn = (ConnectionImpl) getJdbcConnection(properties); + Statement adminStmt = adminConn.createStatement()) { + try { + adminStmt.execute("CREATE ROLE " + String.join(", ", roles)); + adminStmt.execute("CREATE USER " + user + " IDENTIFIED BY '" + userPass + "'"); + waitForAllReplicas("roles", "name IN (" + roleNames + ")", roles.size()); + waitForAllReplicas("users", "name = '" + user + "'", 1); + adminStmt.execute("GRANT " + String.join(", ", roles) + " TO " + user); + adminStmt.execute("SET DEFAULT ROLE NONE TO " + user); + waitForAllReplicas("role_grants", "user_name = '" + user + "' AND granted_role_is_default = 0", + roles.size()); + + Properties info = new Properties(); + info.setProperty("user", user); + info.setProperty("password", userPass); + info.setProperty(DriverProperties.SQL_PARSER.getKey(), parser.name()); + try (ConnectionImpl conn = new ConnectionImpl(getEndpointString(), info)) { + GenericRecord dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 0); + + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE " + role1); + } - Properties info = new Properties(); - info.setProperty("user", "some_user"); - info.setProperty("password", userPass); - info.setProperty(DriverProperties.SQL_PARSER.getKey(), parser.name()); - try (ConnectionImpl conn = new ConnectionImpl(getEndpointString(), info)) { - GenericRecord dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 0); + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 1); + assertEquals(dataRecord.getList(1).get(0), role1); - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE role1"); - } - - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 1); - assertEquals(dataRecord.getList(1).get(0), "role1"); + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE " + role2); + } - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE role2"); - } + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 1); + assertEquals(dataRecord.getList(1).get(0), role2); - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 1); - assertEquals(dataRecord.getList(1).get(0), "role2"); + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE NONE"); + } - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE NONE"); - } + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 0); - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 0); + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE \"" + role1 + "\",\"" + role2 + "\""); + } - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE \"role1\",\"role2\""); - } + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 2); + assertEquals(dataRecord.getList(1).get(0), role1); + assertEquals(dataRecord.getList(1).get(1), role2); - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 2); - assertEquals(dataRecord.getList(1).get(0), "role1"); - assertEquals(dataRecord.getList(1).get(1), "role2"); + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE \"" + role1 + "\",\"" + role2 + "\",\"" + role3 + "\""); + } - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE \"role1\",\"role2\",\"role3\""); - } + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 3); + assertEquals(dataRecord.getList(1).get(0), role1); + assertEquals(dataRecord.getList(1).get(1), role2); + assertEquals(dataRecord.getList(1).get(2), role3); + } - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 3); - assertEquals(dataRecord.getList(1).get(0), "role1"); - assertEquals(dataRecord.getList(1).get(1), "role2"); - assertEquals(dataRecord.getList(1).get(2), "role3"); - } + Properties disableSavingRoles = new Properties(); + disableSavingRoles.setProperty("user", user); + disableSavingRoles.setProperty("password", userPass); + disableSavingRoles.setProperty(DriverProperties.REMEMBER_LAST_SET_ROLES.getKey(), "false"); + disableSavingRoles.setProperty(DriverProperties.SQL_PARSER.getKey(), parser.name()); + try (ConnectionImpl conn = new ConnectionImpl(getEndpointString(), disableSavingRoles)) { + GenericRecord dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 0); - Properties disableSavingRoles = new Properties(); - disableSavingRoles.setProperty("user", "some_user"); - disableSavingRoles.setProperty("password", userPass); - disableSavingRoles.setProperty(DriverProperties.REMEMBER_LAST_SET_ROLES.getKey(), "false"); - disableSavingRoles.setProperty(DriverProperties.SQL_PARSER.getKey(), parser.name()); - try (ConnectionImpl conn = new ConnectionImpl(getEndpointString(), disableSavingRoles)) { - GenericRecord dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 0); + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE " + role1); + } - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE role1"); - } + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 0); - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 0); + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE " + role2); + } - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE role2"); - } + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 0); - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 0); + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE NONE"); + } - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE NONE"); - } + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 0); - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 0); + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE \"" + role1 + "\",\"" + role2 + "\""); + } - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE \"role1\",\"role2\""); - } + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 0); - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 0); + try (Statement stmt = conn.createStatement()) { + stmt.execute("SET ROLE \"" + role1 + "\",\"" + role2 + "\",\"" + role3 + "\""); + } - try (Statement stmt = conn.createStatement()) { - stmt.execute("SET ROLE \"role1\",\"role2\",\"role3\""); + dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); + assertEquals(dataRecord.getList(1).size(), 0); + } + } finally { + adminStmt.execute("DROP USER IF EXISTS " + user); + adminStmt.execute("DROP ROLE IF EXISTS " + String.join(", ", roles)); } - - dataRecord = conn.getClient().queryAll("SELECT currentRoles()").get(0); - assertEquals(dataRecord.getList(1).size(), 0); } } @@ -1359,18 +1371,24 @@ public void testDDLStatements() throws Exception { if (isCloud()) { return; // skip because we do not want to create extra on cloud instance } + String user = "user011_" + RandomStringUtils.insecure().next(8, true, true); try (Connection conn = getJdbcConnection()) { try (Statement stmt = conn.createStatement()) { - Assert.assertFalse(stmt.execute("CREATE USER IF NOT EXISTS 'user011' IDENTIFIED BY 'password'")); + try { + Assert.assertFalse(stmt.execute("CREATE USER IF NOT EXISTS '" + user + "' IDENTIFIED BY 'password'")); + waitForAllReplicas("users", "name = '" + user + "'", 1); - try (ResultSet rs = stmt.executeQuery("SHOW USERS")) { - boolean found = false; - while (rs.next()) { - if (rs.getString("name").equals("user011")) { - found = true; + try (ResultSet rs = stmt.executeQuery("SHOW USERS")) { + boolean found = false; + while (rs.next()) { + if (rs.getString("name").equals(user)) { + found = true; + } } + Assert.assertTrue(found); } - Assert.assertTrue(found); + } finally { + stmt.execute("DROP USER IF EXISTS '" + user + "'"); } } } diff --git a/jdbc-v2/src/test/java/com/clickhouse/jdbc/metadata/ShowStatementDatabaseMetaDataTest.java b/jdbc-v2/src/test/java/com/clickhouse/jdbc/metadata/ShowStatementDatabaseMetaDataTest.java index 73674f3cb..8cc9a660f 100644 --- a/jdbc-v2/src/test/java/com/clickhouse/jdbc/metadata/ShowStatementDatabaseMetaDataTest.java +++ b/jdbc-v2/src/test/java/com/clickhouse/jdbc/metadata/ShowStatementDatabaseMetaDataTest.java @@ -211,22 +211,23 @@ public void testGetColumnsForReadonlyUserWithColumnGrants(String flag, List Date: Mon, 2 Nov 2026 14:05:31 -0800 Subject: [PATCH 2/2] Added waiting data to apear on replica. --- .../client/BaseIntegrationTest.java | 49 ++------ .../client/ClickHouseServerForTest.java | 115 ++++++++++++------ .../com/clickhouse/client/ClientTests.java | 14 ++- .../clickhouse/client/HttpTransportTests.java | 9 +- .../clickhouse/jdbc/JdbcIntegrationTest.java | 24 ++++ 5 files changed, 127 insertions(+), 84 deletions(-) diff --git a/clickhouse-client/src/test/java/com/clickhouse/client/BaseIntegrationTest.java b/clickhouse-client/src/test/java/com/clickhouse/client/BaseIntegrationTest.java index c1a0d72a0..0ce4ff898 100644 --- a/clickhouse-client/src/test/java/com/clickhouse/client/BaseIntegrationTest.java +++ b/clickhouse-client/src/test/java/com/clickhouse/client/BaseIntegrationTest.java @@ -1,11 +1,9 @@ package com.clickhouse.client; import com.clickhouse.client.config.ClickHouseClientOption; -import org.testng.Assert; import org.testng.annotations.AfterTest; import org.testng.annotations.BeforeTest; -import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.Locale; @@ -15,9 +13,6 @@ * Base class for all integration tests. */ public abstract class BaseIntegrationTest { - private static final int REPLICA_CHECK_ATTEMPTS = 100; - private static final long REPLICA_CHECK_INTERVAL_MS = 100L; - @BeforeTest(groups = {"integration"}) public static void setupClickHouseContainer() { ClickHouseServerForTest.beforeSuite(); @@ -98,44 +93,16 @@ protected static String onCluster() { * Does nothing when the tests do not run on a cluster. */ protected static void waitForAllReplicas(String table, String condition, int expectedRows) { - if (!isCluster()) { - return; - } - String sql = "SELECT hostName() AS host, countIf(" + condition + ") AS matching_rows FROM clusterAllReplicas('" - + ClusterClickHouseContainerStarter.CLUSTER_NAME + "', system." + table + ") GROUP BY host"; - String state = ""; - for (int attempt = 1; attempt <= REPLICA_CHECK_ATTEMPTS; attempt++) { - try { - state = ClickHouseServerForTest.queryForResult(sql).trim(); - if (hasRowsOnAllReplicas(state, expectedRows)) { - return; - } - } catch (IOException e) { - state = e.getMessage(); - } - try { - Thread.sleep(REPLICA_CHECK_INTERVAL_MS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - break; - } - } - Assert.fail("Expected " + expectedRows + " rows matching [" + condition + "] in system." + table + " on all " - + ClusterClickHouseContainerStarter.NODE_COUNT + " replicas, but got [host, rows]:\n" + state); + ClickHouseServerForTest.waitForAllReplicas(table, condition, expectedRows); } - private static boolean hasRowsOnAllReplicas(String result, int expectedRows) { - String[] lines = result.split("\n"); - if (lines.length != ClusterClickHouseContainerStarter.NODE_COUNT) { - return false; - } - for (String line : lines) { - String[] columns = line.split("\t"); - if (columns.length != 2 || !columns[1].equals(String.valueOf(expectedRows))) { - return false; - } - } - return true; + /** + * Same check as {@link #waitForAllReplicas(String, String, int)}. {@code query} runs the + * replica SQL with the caller's client and returns one matching-row count per replica host. + */ + protected static void waitForAllReplicas(String table, String condition, int expectedRows, + ClickHouseServerForTest.ReplicaQuery query) { + ClickHouseServerForTest.waitForAllReplicas(table, condition, expectedRows, query); } protected boolean createDatabase(String dbName) { diff --git a/clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseServerForTest.java b/clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseServerForTest.java index 4e452d087..7567c12f2 100644 --- a/clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseServerForTest.java +++ b/clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseServerForTest.java @@ -4,22 +4,24 @@ import org.slf4j.LoggerFactory; import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.Network; +import org.testng.Assert; import org.testng.annotations.AfterSuite; import org.testng.annotations.BeforeSuite; import com.clickhouse.client.ClickHouseTestEnvironment.ContainerMode; +import com.clickhouse.data.ClickHouseRecord; -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.InputStream; import java.io.OutputStream; import java.net.HttpURLConnection; import java.net.InetSocketAddress; import java.net.URL; import java.nio.charset.StandardCharsets; -import java.util.Locale; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Base64; import java.util.Collections; +import java.util.List; +import java.util.Locale; import java.util.Map; /** @@ -36,6 +38,8 @@ @SuppressWarnings("squid:S2187") public class ClickHouseServerForTest { private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseServerForTest.class); + private static final int REPLICA_CHECK_ATTEMPTS = 100; + private static final long REPLICA_CHECK_INTERVAL_MS = 100L; private static final Network network; private static final ClickHouseTestEnvironment environment; @@ -283,47 +287,88 @@ public static boolean runQuery(String sql) { } /** - * Runs a query once and returns the response body (TabSeparated). - * - * @throws IOException when the request fails or the server returns an error + * Waits until each replica of the test cluster returns {@code expectedRows} rows of + * {@code system.
} that match {@code condition}. Does nothing outside cluster mode. + * Uses {@link ClickHouseClient}, which resolves to the HTTP client when that module is on the classpath. */ - public static String queryForResult(String sql) throws IOException { + public static void waitForAllReplicas(String table, String condition, int expectedRows) { + if (!isCluster()) { + return; + } ClickHouseNode server = getClickHouseNode(ClickHouseProtocol.HTTP, isCloud(), ClickHouseNode.builder().build()); - byte[] postData = sql.getBytes(StandardCharsets.UTF_8); - String authorization = Base64.getEncoder() - .encodeToString((getUsername() + ":" + getPassword()).getBytes(StandardCharsets.UTF_8)); - HttpURLConnection httpConn = (HttpURLConnection) new URL(server.getBaseUri()).openConnection(); - try { - httpConn.setConnectTimeout(10000); - httpConn.setReadTimeout(10000); - httpConn.setRequestMethod("POST"); - httpConn.setDoOutput(true); - httpConn.setRequestProperty("Authorization", "Basic " + authorization); - httpConn.setFixedLengthStreamingMode(postData.length); - try (OutputStream out = httpConn.getOutputStream()) { - out.write(postData, 0, postData.length); - } + try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP)) { + waitForAllReplicas(table, condition, expectedRows, sql -> matchingRowCounts(client, server, sql)); + } + } - boolean ok = httpConn.getResponseCode() == HttpURLConnection.HTTP_OK; - ByteArrayOutputStream body = new ByteArrayOutputStream(); - try (InputStream in = ok ? httpConn.getInputStream() : httpConn.getErrorStream()) { - if (in != null) { - byte[] buffer = new byte[4096]; - for (int read; (read = in.read(buffer)) != -1; ) { - body.write(buffer, 0, read); - } + /** + * Same check as {@link #waitForAllReplicas(String, String, int)}. {@code query} runs the + * replica SQL and returns one matching-row count per replica host. + */ + public static void waitForAllReplicas(String table, String condition, int expectedRows, ReplicaQuery query) { + if (!isCluster()) { + return; + } + String sql = "SELECT hostName() AS host, toInt64(countIf(" + condition + ")) AS matching_rows" + + " FROM clusterAllReplicas('" + ClusterClickHouseContainerStarter.CLUSTER_NAME + "', system." + table + + ") GROUP BY host"; + String state = ""; + for (int attempt = 1; attempt <= REPLICA_CHECK_ATTEMPTS; attempt++) { + try { + long[] counts = query.matchingRowCounts(sql); + state = Arrays.toString(counts); + if (hasRowsOnAllReplicas(counts, expectedRows)) { + return; } + } catch (Exception e) { + state = e.getMessage(); } - String result = new String(body.toByteArray(), StandardCharsets.UTF_8); - if (!ok) { - throw new IOException("HTTP " + httpConn.getResponseCode() + ": " + result); + try { + Thread.sleep(REPLICA_CHECK_INTERVAL_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + Assert.fail("Expected " + expectedRows + " rows matching [" + condition + "] in system." + table + " on all " + + ClusterClickHouseContainerStarter.NODE_COUNT + " replicas, but got " + state); + } + + private static long[] matchingRowCounts(ClickHouseClient client, ClickHouseNode server, String sql) + throws ClickHouseException { + try (ClickHouseResponse response = client.read(server).query(sql).executeAndWait()) { + List counts = new ArrayList<>(); + for (ClickHouseRecord record : response.records()) { + counts.add(record.getValue(1).asLong()); + } + long[] result = new long[counts.size()]; + for (int i = 0; i < result.length; i++) { + result[i] = counts.get(i); } return result; - } finally { - httpConn.disconnect(); } } + private static boolean hasRowsOnAllReplicas(long[] counts, int expectedRows) { + if (counts.length != ClusterClickHouseContainerStarter.NODE_COUNT) { + return false; + } + for (long count : counts) { + if (count != expectedRows) { + return false; + } + } + return true; + } + + /** + * Runs the replica-check SQL from {@link #waitForAllReplicas(String, String, int, ReplicaQuery)}. + */ + @FunctionalInterface + public interface ReplicaQuery { + long[] matchingRowCounts(String sql) throws Exception; + } + private static Endpoint resolve(ClickHouseProtocol protocol, int port, boolean applyOverride) { if (starter != null) { int containerPort = (port == ClickHouseProtocol.HTTP.getDefaultPort()) diff --git a/client-v2/src/test/java/com/clickhouse/client/ClientTests.java b/client-v2/src/test/java/com/clickhouse/client/ClientTests.java index e0c7c5d0c..97f3603c6 100644 --- a/client-v2/src/test/java/com/clickhouse/client/ClientTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/ClientTests.java @@ -466,11 +466,14 @@ private static void createUserWithRoles(Client client, String user, String ident final String roleNames = "'" + Strings.join("','", roles) + "'"; client.execute("CREATE ROLE " + rolesList).get().close(); client.execute("CREATE USER " + user + " " + identification).get().close(); - waitForAllReplicas("roles", "name IN (" + roleNames + ")", roles.length); - waitForAllReplicas("users", "name = '" + user + "'", 1); + waitForAllReplicas("roles", "name IN (" + roleNames + ")", roles.length, + sql -> client.queryAll(sql).stream().mapToLong(row -> row.getLong("matching_rows")).toArray()); + waitForAllReplicas("users", "name = '" + user + "'", 1, + sql -> client.queryAll(sql).stream().mapToLong(row -> row.getLong("matching_rows")).toArray()); client.execute("GRANT " + rolesList + " TO " + user).get().close(); - waitForAllReplicas("role_grants", "user_name = '" + user + "' AND granted_role_name IN (" + roleNames + ")", - roles.length); + waitForAllReplicas("role_grants", + "user_name = '" + user + "' AND granted_role_name IN (" + roleNames + ")", roles.length, + sql -> client.queryAll(sql).stream().mapToLong(row -> row.getLong("matching_rows")).toArray()); } @DataProvider(name = "clientSessionRoles") @@ -524,7 +527,8 @@ public void testRuntimeCredentialChange() throws Exception { adminClient.execute("DROP USER IF EXISTS " + user2).get().close(); adminClient.execute("CREATE USER " + user1 + " IDENTIFIED BY '" + password1 + "'").get().close(); adminClient.execute("CREATE USER " + user2 + " IDENTIFIED BY '" + password2 + "'").get().close(); - waitForAllReplicas("users", "name IN ('" + user1 + "', '" + user2 + "')", 2); + waitForAllReplicas("users", "name IN ('" + user1 + "', '" + user2 + "')", 2, + sql -> adminClient.queryAll(sql).stream().mapToLong(row -> row.getLong("matching_rows")).toArray()); try (Client userClient = newClient().setUsername(user1).setPassword(password1).build()) { List firstResponse = userClient.queryAll("SELECT currentUser() AS user"); diff --git a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java index 8a32fcd9b..ffb5730fb 100644 --- a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java @@ -1062,7 +1062,8 @@ public void testSSLAuthentication() throws Exception { try (CommandResponse resp = adminClient.execute("CREATE USER " + user + " IDENTIFIED WITH ssl_certificate CN 'some_user'").get()) { } - waitForAllReplicas("users", "name = '" + user + "'", 1); + waitForAllReplicas("users", "name = '" + user + "'", 1, + sql -> adminClient.queryAll(sql).stream().mapToLong(row -> row.getLong("matching_rows")).toArray()); try (QueryResponse resp = client.query("SELECT 1").get()) { Assert.assertEquals(resp.getReadRows(), 1); @@ -1091,7 +1092,8 @@ public void testPasswordAuthentication(String identifyWith, String identifyBy, b } catch (Exception e) { Assert.fail("Failed on setup", e); } - waitForAllReplicas("users", "name = '" + user + "'", 1); + waitForAllReplicas("users", "name = '" + user + "'", 1, + sql -> adminClient.queryAll(sql).stream().mapToLong(row -> row.getLong("matching_rows")).toArray()); try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) .setUsername(user) @@ -1157,7 +1159,8 @@ public void testAuthHeaderIsKeptFromUser() throws Exception { } catch (Exception e) { Assert.fail("Failed on setup", e); } - waitForAllReplicas("users", "name = '" + user + "'", 1); + waitForAllReplicas("users", "name = '" + user + "'", 1, + sql -> adminClient.queryAll(sql).stream().mapToLong(row -> row.getLong("matching_rows")).toArray()); try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false) .setUsername(user) diff --git a/jdbc-v2/src/test/java/com/clickhouse/jdbc/JdbcIntegrationTest.java b/jdbc-v2/src/test/java/com/clickhouse/jdbc/JdbcIntegrationTest.java index 0361f28d2..7b66a95fb 100644 --- a/jdbc-v2/src/test/java/com/clickhouse/jdbc/JdbcIntegrationTest.java +++ b/jdbc-v2/src/test/java/com/clickhouse/jdbc/JdbcIntegrationTest.java @@ -1,9 +1,12 @@ package com.clickhouse.jdbc; import com.clickhouse.client.BaseIntegrationTest; +import com.clickhouse.client.ClickHouseNode; import com.clickhouse.client.ClickHouseProtocol; import com.clickhouse.client.ClickHouseServerForTest; +import com.clickhouse.client.api.Client; import com.clickhouse.client.api.ClientConfigProperties; +import com.clickhouse.client.api.enums.Protocol; import com.clickhouse.client.api.internal.ServerSettings; import com.clickhouse.data.ClickHouseVersion; import com.clickhouse.logging.Logger; @@ -109,4 +112,25 @@ protected String getServerVersion() { protected boolean isVersionMatch(String versionExpression) { return ClickHouseVersion.of(getServerVersion()).check(versionExpression); } + + /** + * Uses a client-v2 {@link Client} for the replica check. The v1 {@link com.clickhouse.client.ClickHouseClient} + * implementation is not on this module's classpath. + */ + protected static void waitForAllReplicas(String table, String condition, int expectedRows) { + if (!isCluster()) { + return; + } + ClickHouseNode node = ClickHouseServerForTest.getClickHouseNode(ClickHouseProtocol.HTTP, + ClickHouseServerForTest.isCloud(), ClickHouseNode.builder().build()); + try (Client client = new Client.Builder() + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), ClickHouseServerForTest.isCloud()) + .setUsername(ClickHouseServerForTest.getUsername()) + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .build()) { + BaseIntegrationTest.waitForAllReplicas(table, condition, expectedRows, + sql -> client.queryAll(sql).stream().mapToLong(row -> row.getLong("matching_rows")).toArray()); + } + } }