Extend query config with query.masked config with Type.PASSWORD for secure query masking #1593

KGaneshDatta · 2025-11-12T11:07:24Z

Problem

Introduced an internal query.masked configuration of Type.Password to ensure secure logging by masking sensitive values and preventing accidental exposure.
Enhanced the existing validation to enforce that the connector operates exclusively in either table.mode or query.mode.
Added the Retry Logic to fail the connector in case non-retriable SQLExceptions.
Fixed the breaking connector validations in query mode with the new tableFilteringOptions introduced in 10.8.5. Issue

Solution

Does this solution apply anywhere else?

yes
no

If yes, where?

Test Strategy

Testing done:

Unit tests
Integration tests
System tests
Manual tests

Release Plan

…ecure query masking

confluent-cla-assistant · 2025-11-12T11:07:36Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

KGaneshDatta · 2025-11-18T04:44:30Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceConnectorConfig.java

+    orderInGroup = defineIncrementTimestampConfigs(config, orderInGroup);
+    orderInGroup = defineQueryAndQuoteConfigs(config, orderInGroup);
+    defineTransactionAndRetryConfigs(config, orderInGroup);
+  }


Extend this Options method, because of falling to checkstyle error of exceeding 128 lines in a method.

KGaneshDatta · 2025-11-18T04:46:25Z

src/main/java/io/confluent/connect/jdbc/validation/JdbcSourceConnectorValidation.java

+
+    return true;
+  }
+


Added a validation in order to limit using of either of query or query.masked config.

KGaneshDatta · 2025-11-18T04:47:32Z

src/main/java/io/confluent/connect/jdbc/validation/JdbcSourceConnectorValidation.java

+      String msg =
+          "Do not specify table filtering configs with 'query' or 'query.masked'. "
+              + "Remove table.whitelist / table.blacklist / table.include.list / "
+              + "table.exclude.list.";


Either of table.filtering.options or query should be provided as the connector would work in table.mode or query.mode and not both.

akanimesh7

Few things @KGaneshDatta

Since query was never exposed on cloud before this, we should perform some manual testing to understand/validate the behaviour. (How frequently will the query be executed, What if the query is taking too long to exceute ? do we have any observability into this ? Can multiple queries be provided (select * from table1;select* from table2;))
If user has specified both table include/exclude list and query, we don't know which way they wanna go. The validation exception message currently instructs them to remove table include/exclude configs. Which might be other way round also.

akanimesh7 · 2025-11-18T05:07:00Z

src/main/java/io/confluent/connect/jdbc/validation/JdbcSourceConnectorValidation.java

+  private boolean hasAnyQueryConfig() {
+    String query = config.getString(JdbcSourceConnectorConfig.QUERY_CONFIG);
+    Password queryMasked =
+        config.getPassword(JdbcSourceConnectorConfig.QUERY_MASKED_CONFIG);


If you make config.getQuery return optional, then this hasAnyQueryConfig method will not be needed.

…on to allow only select queries in query configs

…masked has been set

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceConnectorConfig.java

akanimesh7 · 2025-12-01T06:07:51Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

      } else if (mode.equals(JdbcSourceTaskConfig.MODE_INCREMENTING)) {
        tableQueue.add(
            new TimestampIncrementingTableQuerier(
+                config,


config is unused in the passed class. What's the need of this new param in the constructor ?

The config object has been used in TableQuerier class in order to check for query config has been set or not. As, TimestampIncrementingTableQuerier is referencing TableQuerier as super class, it is definitely required to include all the arguments.

akanimesh7 · 2025-12-01T06:08:17Z

src/main/java/io/confluent/connect/jdbc/source/TableQuerier.java

    this.lastUpdate = 0;
    this.suffix = suffix;
    this.attemptedRetries = 0;
+    this.shouldTrimSensitiveLogs = config.isQueryMasked();


Directly pass the config.isQueryMasked value, rather than whole config object

Yes, agree. But if we pass the config object, then it would be used for all other methods in future instead of each argument for each usecase.

src/main/java/io/confluent/connect/jdbc/util/LogUtil.java

src/main/java/io/confluent/connect/jdbc/source/TableQuerierProcessor.java

akanimesh7 · 2025-12-01T06:25:26Z

src/main/java/io/confluent/connect/jdbc/source/TableQuerierProcessor.java

                                           TableQuerier querier, SQLNonTransientException sqle) {
+    SQLException trimmedException = shouldTrimSensitiveLogs
+              ? LogUtil.trimSensitiveData(sqle) : sqle;
+    String querierLog = getQuerierLogString(querier);


rather than redacting the whole querier, simply redact the query member var in the toString for querier object.

src/main/java/io/confluent/connect/jdbc/source/TableQuerierProcessor.java

…querier object

sonarqube-confluent · 2025-12-01T10:09:34Z

Quality Gate failed

Failed conditions
0.0% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

akanimesh7

Missing validation -- If mode is query and timestamp.initial is provided validation passes.

akanimesh7 · 2025-12-01T13:07:37Z

src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java

    );
  }

  protected Map<String, Object> computeInitialOffset(


refactor -- don't enter this function if we are in query mode. Offsets are always null for query mode. And hence no need to redact here since it will always be table name.

akanimesh7 · 2025-12-01T13:11:24Z

src/main/java/io/confluent/connect/jdbc/source/TableQuerier.java

    }
  }

+  protected String getQuerierLogString(String query) {


Can be shortened as suggested previously

Why do we need this function, can't we simply replace this function call with LogUtil.sensitiveLog(shouldTrimSensitiveLogs, query) ?

akanimesh7 · 2025-12-01T13:24:42Z

src/main/java/io/confluent/connect/jdbc/validation/JdbcSourceConnectorValidation.java


-}
+  /** Validate that provided query strings start with a SELECT statement. */
+  private boolean validateSelectStatement(String statement, String configKey) {


String configKey is unncessary. It can be removed. There is only one possible value -- query

akanimesh7 · 2025-12-01T13:25:13Z

pom.xml

            <version>1.7</version>
+        </dependency>
+        <dependency>
+          <groupId>org.apache.commons</groupId>


Extend query config with query.masked config with Type.PASSWORD for s…

5198b37

…ecure query masking

KGaneshDatta requested a review from a team as a code owner November 12, 2025 11:07

KGaneshDatta added 4 commits November 13, 2025 13:35

Fix checkstyle errors and added tests

bb93366

Making query masked config internal and fixed validation

330e9fe

Fix UTs with Validation errors

b8d712a

Update the documentation of new config

3a22709

KGaneshDatta commented Nov 18, 2025

View reviewed changes

akanimesh7 reviewed Nov 18, 2025

View reviewed changes

KGaneshDatta added 6 commits November 21, 2025 15:00

Update getQuery() method to return optional

9330777

Minor checkstyle fixes

48350ca

Fixed Validation with query mode

a202be6

Added functionality to fail tasks when non-retriable exceptions

0081562

Redact the sensitive values from logs and exceptions and add validati…

7eb279a

…on to allow only select queries in query configs

Remove RetryUtils and add condition to redact values only when query.…

34a0807

…masked has been set

akanimesh7 reviewed Dec 1, 2025

View reviewed changes

KGaneshDatta added 2 commits December 1, 2025 13:17

Minor changes in comments and code

d1dfc62

Removed config object in Tablequerier and masked only query value in …

226373d

…querier object

akanimesh7 reviewed Dec 1, 2025

View reviewed changes

Extend query config with query.masked config with Type.PASSWORD for secure query masking #1593

Are you sure you want to change the base?

Extend query config with query.masked config with Type.PASSWORD for secure query masking #1593

Uh oh!

Conversation

KGaneshDatta commented Nov 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Problem

Solution

Does this solution apply anywhere else?

If yes, where?

Test Strategy

Testing done:

Release Plan

Uh oh!

confluent-cla-assistant bot commented Nov 12, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

akanimesh7 left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

sonarqube-confluent bot commented Dec 1, 2025

Quality Gate failed

Uh oh!

akanimesh7 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

KGaneshDatta commented Nov 12, 2025 •

edited

Loading

akanimesh7 left a comment •

edited

Loading