public QueryDTO.ClusterQueryCommanderResult execute(
        String clusterId,
        QueryDTO.ClusterQueryCommanderArgs args
    ) {
        CqlSession session = cqlSessionFactory.get(clusterId);
        if (useKeyspaceQueryCommandNotSupportWithSession(session) && StringUtils.isNotBlank(args.getKeyspace())) {
            throw new NotSupportedCassandraVersionException("It is available in Cassandra version 4.0 and later");
        }

        SimpleStatementBuilder simpleBuilder = SimpleStatement.builder(args.getQuery())
            .setPageSize(args.getPageSize())                    // 10 per pages
            .setTimeout(Duration.ofSeconds(args.getTimeoutSeconds()))  // 3s timeout
            .setPagingState(StringUtils.isNotBlank(args.getCursor()) ? Bytes.fromHexString(args.getCursor()) : null)
            .setConsistencyLevel(args.getConsistencyLevel())
            .setTracing(args.isTrace());

        if (StringUtils.isNotBlank(args.getKeyspace())) {
            simpleBuilder.setKeyspace(args.getKeyspace());
        }

        SimpleStatement statement = simpleBuilder.build();

        ResultSet resultSet = session.execute(statement);
        ColumnDefinitions definitions = resultSet.getColumnDefinitions();

        ByteBuffer pagingStateAsBytes = resultSet.getExecutionInfo().getPagingState();

        QueryDTO.ClusterQueryCommanderResult.ClusterQueryCommanderResultBuilder builder =
            QueryDTO.ClusterQueryCommanderResult.builder()
                .wasApplied(resultSet.wasApplied())
                .rowHeader(CassdioColumnDefinition.makes(definitions))
                .rows(convertRows(session, resultSet))
                .nextCursor(pagingStateAsBytes != null ? Bytes.toHexString(pagingStateAsBytes) : null);

        if (args.isTrace()) {
            QueryTrace queryTrace = resultSet.getExecutionInfo().getQueryTrace();

            List<QueryDTO.CassdioQueryTraceEvent> events = new ArrayList<>();
            for (TraceEvent event : queryTrace.getEvents()) {
                events.add(QueryDTO.CassdioQueryTraceEvent.builder()
                    .activity(event.getActivity())
                    .timestamp(event.getTimestamp())
                    .sourceAddress(event.getSourceAddress())
                    .sourceElapsedMicros(event.getSourceElapsedMicros())
                    .threadName(event.getThreadName())
                    .build());
            }

            QueryDTO.CassdioQueryTrace cassdioQueryTrace = QueryDTO.CassdioQueryTrace.builder()
                .tracingId(queryTrace.getTracingId().toString())
                .requestType(queryTrace.getRequestType())
                .coordinatorAddress(queryTrace.getCoordinatorAddress())
                .parameters(queryTrace.getParameters())
                .startedAt(queryTrace.getStartedAt())
                .events(events)
                .build();

            builder.queryTrace(cassdioQueryTrace);
        }

        return builder.build();
    }
@Test
    void queryTest() {
        //GIVEN
        QueryDTO.ClusterQueryCommanderArgs args = QueryDTO.ClusterQueryCommanderArgs.builder()
            .query(String.format("SELECT * FROM %s.%s", keyspaceName, TABLE_NAME))
            .build();

        //WHEN
        QueryDTO.ClusterQueryCommanderResult result = clusterQueryCommander.execute(CLUSTER_ID, args);

        //THEN
        assertThat(result).isNotNull();
        assertThat(result.getRows()).isNotNull();

    }