Improve Handling of Missing Database Key Fields in Pulsar IO JDBC Sink
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Motivation
When using the Pulsar IO JDBC Sink with PostgreSQL and Avro schemas - if the database has a column that does not have a corresponding field in the schema/message, the sink does not work.
I noticed this issue when trying to integrate with a table that has a primary key populated by the db, i.e. using a regular db sequence. The seemingly only workaround is have my app call the db sequence and set the ID in the message before passing it to the sink. This seems a basic use case and would be very useful feature.
### Solution
Improve the JDBC Sink to handle cases where messages lack db fields, by introducing additional configuration options:
Option 1: `missingKeyHandlingMode`
`FAIL` (default): Keep existing behavior, failing when a db field is missing from message.
`IGNORE`: Use db default value.
Option 2: `ignoreKeys`
CSV of message keys to not attempt to map to a db column
### Alternatives
_No response_
### Anything else?
**Current errors**
**Setup**
* using Pulsar 4.0.2
* The db table has an id field defined as: `id integer` - with no constraints.
* The message/schem does not have an id field.
`org.apache.avro.AvroRuntimeException: Not a valid schema field: id
at org.apache.avro.generic.GenericData$Record.get(GenericData.java:282) ~[java-instance.jar:?]
at org.apache.pulsar.client.impl.schema.generic.GenericAvroRecord.getField(GenericAvroRecord.java:48) ~[pulsar-client-original-4.0.2.jar:4.0.2]
at org.apache.pulsar.io.jdbc.BaseJdbcAutoSchemaSink.lambda$createMutation$1(BaseJdbcAutoSchemaSink.java:146) ~[pulsar-io-jdbc-core-4.0.2.jar:?]`
I did test `excludeNonDeclaredFields` set to true, which gives error also:
`2025-02-08T10:02:47,417+0000 [pool-5-thread-1] ERROR org.apache.pulsar.io.jdbc.JdbcAbstractSink - Got exception No value specified for parameter 1. after 0 ms, failing 1 messages
org.postgresql.util.PSQLException: No value specified for parameter 1.
at org.postgresql.core.v3.SimpleParameterList.checkAllParametersSet(SimpleParameterList.java:339) ~[postgresql-42.5.5.jar:42.5.5]
at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:340) ~[postgresql-42.5.5.jar:42.5.5]
at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:496) ~[postgresql-42.5.5.jar:42.5.5]`
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
Research direction
Start with BaseJdbcAutoSchemaSink.createMutation, where the reported GenericAvroRecord field lookup fails, and reproduce the PostgreSQL/Avro setup described in the issue. Trace how excludeNonDeclaredFields supplies statement parameters, then verify that the chosen missing-field behavior uses database defaults or ignores configured keys without parameter errors.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, postgresql
- Domain
- data-engineering, databases
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 38/100