apache / apache/pulsar

2.8.0 sink can no longer process batched records from Python producer

Open
#11,288 9 comments 0 reactions 0 assignees View on GitHub
lifecycle/stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

**Describe the bug**
I have a simple Python producer that has been sending messages to Pulsar since 2.6.1 with `batching_enabled=True`. These messages are processed an instance of the built-in InfluxDB sink connector. Since upgrading to 2.8.0, I receive this warning and the records are not processed:

```
12:53:20.020 [pool-2-thread-1] WARN org.apache.pulsar.io.influxdb.BatchSink - Record flush thread was exception
org.apache.pulsar.client.api.SchemaSerializationException: measurement is a required field.
at org.apache.pulsar.io.influxdb.v1.InfluxDBGenericRecordSink.buildPoint(InfluxDBGenericRecordSink.java:55) ~[pulsar-io-influxdb-2.8.0.nar-unpacked/:?]
at org.apache.pulsar.io.influxdb.v1.InfluxDBGenericRecordSink.buildPoint(InfluxDBGenericRecordSink.java:41) ~[pulsar-io-influxdb-2.8.0.nar-unpacked/:?]
at org.apache.pulsar.io.influxdb.BatchSink.flush(BatchSink.java:85) ~[pulsar-io-influxdb-2.8.0.nar-unpacked/:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_292]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) [?:1.8.0_292]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_292]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) [?:1.8.0_292]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_292]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_292]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_292]
```

My messages do have a simple schema required by the sink and `measurement` (as seen above) is defined and present in them. More importantly, messages sent with `batching_enabled=False` continue to work fine.

I added some additional logging to the sink and was able to observe that when `batching_enabled=False`, records seen in `org.apache.pulsar.io.influxdb.BatchSink.flush()` are of class `GenericJsonRecord`, but when batching is set to True, they are of `GenericObjectWrapper`. In 2.7.x, they are `GenericJsonRecord` in both cases.

I also tested using Avro instead of JSON, with the same behavior.

**To Reproduce**
Steps to reproduce the behavior:
1. Create a simple Python producer:
```
from pulsar import Client
from pulsar.schema import JsonSchema, Record, String

class Test_Schema(Record):
measurement = String(required=True)
field1 = String()

client = Client('pulsar://localhost:6650')
producer = client.create_producer(topic='mytopic', schema=JsonSchema(Test_Schema), batching_enabled=True)
producer.send(Test_Schema(measurement='foo', field1='bar'))
```
2. Connect InfluxDB sink to 'mytopic' in localrun mode. I am using this config:
```
configs:
influxdbUrl: "http://influxdb:8086"
database: "mydb"
consistencyLevel: "ONE"
logLevel: "NONE"
retentionPolicy: "autogen"
batchTimeMs: 1000
batchSize: 100
```
3. Run script
4. See error

**Expected behavior**
No error generated. Message sent to InfluxDB and written to database.

**Additional context**
As mentioned, this same code works without issue on Pulsar 2.6.1 and 2.7.x. The Python client version does not seem to matter either (I tested using 2.6.1 and 2.8.0). It is only during the recent Pulsar upgrade to 2.8.0 that the batch processing no longer works.

Contributor guide

Open the contributing guide

Research direction

Start with org.apache.pulsar.io.influxdb.BatchSink.flush() and InfluxDBGenericRecordSink.buildPoint(), using the reported stack trace locations. Reproduce with the Python JsonSchema producer and batching_enabled=True, then compare the GenericObjectWrapper path with GenericJsonRecord. Done means batched records no longer raise the measurement error and are written to InfluxDB.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, python
Domain
backend, databases, stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Clearly specified
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.