Windows下Kafka Connect Timestamp+Incrementing模式同步数据失败排查

Let's dive right into solving your problem—you've got a Kafka Connect JDBC Source Connector running in timestamp+incrementing mode that won't pull new Oracle data, even though the connector and tasks show as RUNNING, and Bulk mode works fine. The debug logs you shared give away the key issue: the start and end timestamps used in the query are identical (2019-08-23 11:40:47:046), which means the connector thinks there's no new data to sync. This is almost always a timezone mismatch problem.

Here's how to fix it step by step:

1. Align Database and Connector Timezones Properly #

Your Oracle table uses SYS_EXTRACT_UTC(SYSTIMESTAMP) to store UTC time in the regist field, and you've set db.timezone=UTC in the connector config—good start. But your custom query is converting a string field (STRING_TIMESTAMP) to a timestamp, which is risky if that string doesn't explicitly include timezone info.

If STRING_TIMESTAMP stores time in your local timezone (not UTC), the conversion will use Oracle's default timezone, creating a mismatch with the connector's UTC setting. Fix this by explicitly converting the string to UTC in your query:

select NDZ, 
  from_tz(to_timestamp(STRING_TIMESTAMP, 'YYYY-MM-DD HH24:MI:SS.FF6'), 'YOUR_LOCAL_TIMEZONE') 
  at time zone 'UTC' AS TIMESTAMP_COLUMN 
FROM "usu"."mytable"

(Replace YOUR_LOCAL_TIMEZONE with the actual timezone of the string data, e.g., America/New_York)

Even better: skip the string conversion entirely. Since your regist field is already a UTC timestamp, use it directly in your query:

select NDZ, regist AS TIMESTAMP_COLUMN FROM "usu"."mytable"

Then keep timestamp.column.name=TIMESTAMP_COLUMN in your config—this eliminates any conversion-related timezone errors.

2. Fix the Malformed Query in Debug Logs #

Looking at the debug SQL, there's a syntax error: a stray closing parenthesis after FROM myTable:

select ... FROM myTable) WHERE "TIMESTAMP_COLUMN" < ? ...

This could cause the query to silently fail (even if the connector shows RUNNING). To avoid manual query syntax mistakes, ditch the custom query config and let the connector generate the SQL automatically. Use table.whitelist instead:

# Remove the query line, add this instead
table.whitelist=usu.mytable
timestamp.column.name=regist
incrementing.column.name=NDZ

This ensures the connector builds a valid, timezone-aware query every time.

3. Reset Connector Offsets #

Kafka Connect stores the last synced timestamp and incrementing ID in the connect-offsets Kafka topic. If your previous misconfiguration set this offset to a time later than your new data, the connector will ignore existing records. Here's how to reset it:

1. Stop the connector:

   curl -X DELETE http://localhost:8083/connectors/jdbc-conector
   

2. Delete and recreate the connect-offsets topic (note: this resets offsets for all connectors—if you have others, use a Kafka offset management tool to delete only this connector's offsets instead):

   kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic connect-offsets
   kafka-topics.sh --bootstrap-server localhost:9092 --create --topic connect-offsets --partitions 1 --replication-factor 1 --config cleanup.policy=compact
   

3. Re-deploy your corrected connector config.

4. Verify the Incrementing Column #

Double-check that NDZ is a strictly increasing column (like an auto-incrementing primary key). If it's not unique or doesn't increment with each new row, the timestamp+incrementing mode can't reliably track new records.

After making these changes, insert a new test record with INSERT INTO "usu"."mytable" (first_name,last_name) VALUES ('jake','tyler') and run the consumer command again—you should see the data show up in the input-mytable topic.

内容的提问来源于stack exchange，提问作者Lanre