redpanda-data / redpanda-data/benthos
message when deleted inside retry processor, does not apply
Nobody has claimed this yet.
- Dominant language
- Go
- Stars
- 571
- Forks
- 120
- Avg merge
- 2d 1h
- Merged PRs (30d)
- 18
Description
Seeing something abnormal with retry processor, if the message was deleted in one of the retry processors and using a sql_insert output, was not expecting execution of sql_insert. But sql_insert fails with a error. Here is a sample pipeline that fails.
input:
generate:
interval: 1m
mapping: |-
root = { "p1": "test val" }
pipeline:
processors:
- retry:
backoff:
initial_interval: 500ms
max_interval: 10s
max_elapsed_time: 0s
processors:
- log:
message: 'incoming msg ${! content() }'
- mapping: root = deleted()
output:
sql_insert:
driver: postgres
dsn: postgres://${DB_USER}:${DB_PASSWORD}@${DB_HOST}:5432/${DB_NAME}?sslmode=disable
conn_max_idle_time: 1m
table: test_table
columns:
- p1
args_mapping: |
root = [
this.p1
]
init_statement:
CREATE TABLE IF NOT EXISTS test_table (
p1 varchar
)
and console output:
INFO incoming msg {"p1":"test_value"} @service=benthos custom_source=true label="" path=root.pipeline.processors.0.retry.processors.0
INFO Output type sql_insert is now active @service=benthos label="" path=root.output
ERRO Failed to send message to sql_insert: insert statements must have at least one set of values or select clause @service=benthos label="" path=root.output
INFO incoming msg {"p1":"test_value"} @service=benthos custom_source=true label="" path=root.pipeline.processors.0.retry.processors.0
ERRO Failed to send message to sql_insert: insert statements must have at least one set of values or select clause @service=benthos label="" path=root.output
INFO incoming msg {"p1":"test_value"} @service=benthos custom_source=true label="" path=root.pipeline.processors.0.retry.processors.0
Could workaround it by deleting message in a pre-processor to sql_insert, like below:
output:
sql_insert:
driver: postgres
dsn: postgres://${DB_USER}:${DB_PASSWORD}@${DB_HOST}:5432/${DB_NAME}?sslmode=disable
conn_max_idle_time: 1m
table: test_table
columns:
- p1
args_mapping: |
root = [
this.p1
]
init_statement:
CREATE TABLE IF NOT EXISTS test_table (
p1 varchar
)
processors:
- mapping: |-
root = if this == null {
deleted()
}
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Reproduce the supplied pipeline with the retry processor and sql_insert output, then trace how a deleted message moves from the retry processors to the output. Confirm the behavior against the PostgreSQL example; done means a message deleted inside retry processing does not cause sql_insert to execute.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go, postgresql
- Domain
- backend, databases
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100