apache / apache/pulsar

Improve Handling of Missing Database Key Fields in Pulsar IO JDBC Sink

Open
#23,953 0 comments 0 reactions 0 assignees View on GitHub
type/enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.