Code Review Suggestion on allow multiple database from connector.

[abhishekgahlot2]

Hi, I am planning to raise a pr to allow multiple database from topics name. I have gotten this far. would be great more of your help.

Changing client implementation

public ClickHouseHelperClient getClientChc(String databaseName) {
        return new ClickHouseHelperClient.ClickHouseClientBuilder(csc.getHostname(), csc.getPort(), csc.getProxyType(), csc.getProxyHost(), csc.getProxyPort())
        .setDatabase(databaseName)
        .setUsername(csc.getUsername())
        .setPassword(csc.getPassword())
        .sslEnable(csc.isSslEnabled())
        .setTimeout(csc.getTimeout())
        .setRetry(csc.getRetry())
        .build();
    }

Changing database in mutation request

private ClickHouseRequest.Mutation getMutationRequest(ClickHouseClient client, ClickHouseFormat format, String tableName, QueryIdentifier queryId, String databaseName) {
        ClickHouseRequest.Mutation request = client.read(getClientChc(databaseName).getServer())
                .write()
                .table(tableName, queryId.getQueryId())
                .format(format)
                .set("insert_deduplication_token", queryId.getDeduplicationToken());

        for (String clickhouseSetting : csc.getClickhouseSettings().keySet()) {//THIS ASSUMES YOU DON'T ADD insert_deduplication_token
            request.set(clickhouseSetting, csc.getClickhouseSettings().get(clickhouseSetting));
        }

        request.option(ClickHouseClientOption.WRITE_BUFFER_SIZE, 8192);

        return request;
    }

Then change, insert json, insert binary, insert string to fetch topic and get database from topic

        Record first = records.get(0);
        String topic = first.getRecordOffsetContainer().getTopic();
        String[] topicParts = topic0.split("_");

        try (ClickHouseClient client = getClient()) {
            ClickHouseRequest.Mutation request = getMutationRequest(client, ClickHouseFormat.JSONEachRow, table.getName(), queryId, topicParts[0]);
            ClickHouseConfig config = request.getConfig();
            CompletableFuture<ClickHouseResponse> future;

Any ideas if i am doing it correctly or performance bottleneck here?

What i saw so far is the validate schema on start checks if topics and table schema matches with default so that is one blocker.

Thanks in advance