Hello,
Any updates on this bug?
Thanks!
On Fri, Nov 20, 2020 at 11:51 AM Rex Fenley <r...@remind101.com> wrote:
> Btw, this is what our source and sink essentially look like, with some
> columns redacted.
>
> CREATE TABLE source_kafka_data (
> id BIGINT,
> roles ARRAY<STRING NOT NULL>,
> PRIMARY KEY (id) NOT ENFORCED
> ) WITH (
> 'connector' = 'kafka',
> 'topic' = 'topic',
> 'properties.bootstrap.servers' = 'kafka',
> 'properties.group.id' = 'group_id',
> 'properties.auto.offset.reset' = 'earliest',
> 'debezium-json.schema-include' = 'true',
> 'format' = 'debezium-json'
> )
>
>
> CREATE TABLE sink_es_data (
> id BIGINT NOT NULL,
> roles ARRAY<STRING>,
> PRIMARY KEY (id) NOT ENFORCED
> ) WITH (
> 'connector' = 'elasticsearch-7',
> 'hosts' = 'eshost',
> 'index' = 'data',
> 'format' = 'json',
> 'sink.bulk-flush.max-actions' = '8192',
> 'sink.bulk-flush.max-size' = '16mb',
> 'sink.bulk-flush.interval' = '5000',
> 'sink.bulk-flush.backoff.delay' = '1000',
> 'sink.bulk-flush.backoff.max-retries' = '4',
> 'sink.bulk-flush.backoff.strategy' = 'CONSTANT'
> )
>
>
>
> On Thu, Nov 19, 2020 at 7:41 PM Rex Fenley <r...@remind101.com> wrote:
>
>> Thanks!
>>
>> Update: We've confirmed with a test copy of our data now that if we
>> remove all the null values from arrays everything works smoothly and as
>> expected. So this definitely appears to be the culprit.
>>
>> On Thu, Nov 19, 2020 at 6:41 PM Jark Wu <imj...@gmail.com> wrote:
>>
>>> Thanks Rex! This is very helpful. Will check it out later.
>>>
>>>
>>> On Fri, 20 Nov 2020 at 03:02, Rex Fenley <r...@remind101.com> wrote:
>>>
>>>> Below is a highly redacted set of data that should represent the
>>>> problem. As you can see, the "roles" field has "[null]" in it, a null value
>>>> within the array. We also see in our DB corresponding rows like the
>>>> following.
>>>> id | roles
>>>> -----------+------------
>>>> 16867433 | {NULL}
>>>>
>>>> We have confirmed that by not selecting "roles" all data passes through
>>>> without failure on a single operator, but selecting "roles" will eventually
>>>> always fail with java.lang.NullPointerException repeatedly. What is
>>>> odd about this is there is 0 additional stack trace, just the exception, in
>>>> our logs and in Flink UI. We only have INFO logging on, however, other
>>>> exceptions we've encountered in our development have always revealed a
>>>> stack trace.
>>>>
>>>> {
>>>> "schema": {
>>>> "type": "struct",
>>>> "fields": [
>>>> {
>>>> "type": "struct",
>>>> "fields": [
>>>> { "type": "int32", "optional": false, "field": "id" },
>>>> {
>>>> "type": "array",
>>>> "items": { "type": "string", "optional": true },
>>>> "optional": false,
>>>> "field": "roles"
>>>> },
>>>> ],
>>>> "optional": true,
>>>> "name": "db.public.data.Value",
>>>> "field": "before"
>>>> },
>>>> {
>>>> "type": "struct",
>>>> "fields": [
>>>> { "type": "int32", "optional": false, "field": "id" },
>>>> {
>>>> "type": "array",
>>>> "items": { "type": "string", "optional": true },
>>>> "optional": false,
>>>> "field": "roles"
>>>> },
>>>> ],
>>>> "optional": true,
>>>> "name": "db.public.data.Value",
>>>> "field": "after"
>>>> },
>>>> {
>>>> "type": "struct",
>>>> "fields": [
>>>> { "type": "string", "optional": false, "field": "version" },
>>>> { "type": "string", "optional": false, "field": "connector" },
>>>> { "type": "string", "optional": false, "field": "name" },
>>>> { "type": "int64", "optional": false, "field": "ts_ms" },
>>>> {
>>>> "type": "string",
>>>> "optional": true,
>>>> "name": "io.debezium.data.Enum",
>>>> "version": 1,
>>>> "parameters": { "allowed": "true,last,false" },
>>>> "default": "false",
>>>> "field": "snapshot"
>>>> },
>>>> { "type": "string", "optional": false, "field": "db" },
>>>> { "type": "string", "optional": false, "field": "schema" },
>>>> { "type": "string", "optional": false, "field": "table" },
>>>> { "type": "int64", "optional": true, "field": "txId" },
>>>> { "type": "int64", "optional": true, "field": "lsn" },
>>>> { "type": "int64", "optional": true, "field": "xmin" }
>>>> ],
>>>> "optional": false,
>>>> "name": "io.debezium.connector.postgresql.Source",
>>>> "field": "source"
>>>> },
>>>> { "type": "string", "optional": false, "field": "op" },
>>>> { "type": "int64", "optional": true, "field": "ts_ms" },
>>>> {
>>>> "type": "struct",
>>>> "fields": [
>>>> { "type": "string", "optional": false, "field": "id" },
>>>> { "type": "int64", "optional": false, "field": "total_order"
>>>> },
>>>> {
>>>> "type": "int64",
>>>> "optional": false,
>>>> "field": "data_collection_order"
>>>> }
>>>> ],
>>>> "optional": true,
>>>> "field": "transaction"
>>>> }
>>>> ],
>>>> "optional": false,
>>>> "name": "db.public.data.Envelope"
>>>> },
>>>> "payload": {
>>>> "before": null,
>>>> "after": {
>>>> "id": 76704,
>>>> "roles": [null],
>>>> },
>>>> "source": {
>>>> "version": "1.3.0.Final",
>>>> "connector": "postgresql",
>>>> "name": "db",
>>>> "ts_ms": 1605739197360,
>>>> "snapshot": "true",
>>>> "db": "db",
>>>> "schema": "public",
>>>> "table": "data",
>>>> "txId": 1784,
>>>> "lsn": 1305806608,
>>>> "xmin": null
>>>> },
>>>> "op": "r",
>>>> "ts_ms": 1605739197373,
>>>> "transaction": null
>>>> }
>>>> }
>>>>
>>>> cc Brad
>>>>
>>>> On Thu, Nov 19, 2020 at 7:29 AM Dylan Forciea <dy...@oseberg.io> wrote:
>>>>
>>>>> Ah yes, missed the kafka part and just saw the array part. FLINK-19771
>>>>> definitely was solely in the postgres-specific code.
>>>>>
>>>>>
>>>>>
>>>>> Dylan
>>>>>
>>>>>
>>>>>
>>>>> *From: *Jark Wu <imj...@gmail.com>
>>>>> *Date: *Thursday, November 19, 2020 at 9:12 AM
>>>>> *To: *Dylan Forciea <dy...@oseberg.io>
>>>>> *Cc: *Danny Chan <danny0...@apache.org>, Rex Fenley <r...@remind101.com>,
>>>>> Flink ML <user@flink.apache.org>
>>>>> *Subject: *Re: Filter Null in Array in SQL Connector
>>>>>
>>>>>
>>>>>
>>>>> Hi Dylan,
>>>>>
>>>>>
>>>>>
>>>>> I think Rex encountered another issue, because he is using Kafka with
>>>>> Debezium format.
>>>>>
>>>>>
>>>>>
>>>>> Hi Rex,
>>>>>
>>>>>
>>>>>
>>>>> If you can share the json data and the exception stack, that would be
>>>>> helpful!
>>>>>
>>>>>
>>>>>
>>>>> Besides, you can try to enable 'debezium-json.ignore-parse-errors'
>>>>> option [1] to skip the dirty data.
>>>>>
>>>>>
>>>>>
>>>>> Best,
>>>>>
>>>>> Jark
>>>>>
>>>>>
>>>>>
>>>>> [1]:
>>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/debezium.html#debezium-json-ignore-parse-errors
>>>>>
>>>>>
>>>>>
>>>>> On Thu, 19 Nov 2020 at 21:13, Dylan Forciea <dy...@oseberg.io> wrote:
>>>>>
>>>>> Do you mean that the array contains values that are null, or that the
>>>>> entire array itself is null? If it’s the latter, I have an issue written,
>>>>> along with a PR to fix it that has been pending review [1].
>>>>>
>>>>>
>>>>>
>>>>> Regards,
>>>>>
>>>>> Dylan Forciea
>>>>>
>>>>>
>>>>>
>>>>> [1] https://issues.apache.org/jira/browse/FLINK-19771
>>>>>
>>>>>
>>>>>
>>>>> *From: *Danny Chan <danny0...@apache.org>
>>>>> *Date: *Thursday, November 19, 2020 at 2:24 AM
>>>>> *To: *Rex Fenley <r...@remind101.com>
>>>>> *Cc: *Flink ML <user@flink.apache.org>
>>>>> *Subject: *Re: Filter Null in Array in SQL Connector
>>>>>
>>>>>
>>>>>
>>>>> Hi, Fenley ~
>>>>>
>>>>>
>>>>>
>>>>> You are right, parsing nulls of ARRAY field is not supported now, i
>>>>> have logged an issue [1] and would fix it soon ~
>>>>>
>>>>>
>>>>>
>>>>> [1] https://issues.apache.org/jira/browse/FLINK-20234
>>>>>
>>>>>
>>>>>
>>>>> Rex Fenley <r...@remind101.com> 于2020年11月19日周四 下午2:51写道：
>>>>>
>>>>> Hi,
>>>>>
>>>>>
>>>>>
>>>>> I recently discovered some of our data has NULL values arriving in an
>>>>> ARRAY<STRING> column. This column is being consumed by Flink via the Kafka
>>>>> connector Debezium format. We seem to be receiving NullPointerExceptions
>>>>> for when these NULL values in the arrays arrive which restarts the source
>>>>> operator in a loop.
>>>>>
>>>>>
>>>>>
>>>>> Is there any way to not throw or to possibly filter out NULLs in an
>>>>> Array of Strings in Flink?
>>>>>
>>>>>
>>>>>
>>>>> We're somewhat stuck on how to solve this problem, we'd like to be
>>>>> defensive about this on Flink's side.
>>>>>
>>>>>
>>>>>
>>>>> Thanks!
>>>>>
>>>>>
>>>>>
>>>>> (P.S. The exception was not that informative, there may be room for
>>>>> improvement in terms of a richer error message when this happens.)
>>>>>
>>>>>
>>>>> --
>>>>>
>>>>> *Rex Fenley* | Software Engineer - Mobile and Backend
>>>>>
>>>>>
>>>>>
>>>>> *Remind.com* <https://www.remind.com/> | BLOG
>>>>> <http://blog.remind.com/> | FOLLOW US <https://twitter.com/remindhq>
>>>>> | *LIKE US <https://www.facebook.com/remindhq>*
>>>>>
>>>>>
>>>>
>>>> --
>>>>
>>>> Rex Fenley | Software Engineer - Mobile and Backend
>>>>
>>>>
>>>> Remind.com <https://www.remind.com/> | BLOG <http://blog.remind.com/>
>>>> | FOLLOW US <https://twitter.com/remindhq> | LIKE US
>>>> <https://www.facebook.com/remindhq>
>>>>
>>>
>>
>> --
>>
>> Rex Fenley | Software Engineer - Mobile and Backend
>>
>>
>> Remind.com <https://www.remind.com/> | BLOG <http://blog.remind.com/>
>> | FOLLOW US <https://twitter.com/remindhq> | LIKE US
>> <https://www.facebook.com/remindhq>
>>
>
>
> --
>
> Rex Fenley | Software Engineer - Mobile and Backend
>
>
> Remind.com <https://www.remind.com/> | BLOG <http://blog.remind.com/> |
> FOLLOW US <https://twitter.com/remindhq> | LIKE US
> <https://www.facebook.com/remindhq>
>
--
Rex Fenley | Software Engineer - Mobile and Backend
Remind.com <https://www.remind.com/> | BLOG <http://blog.remind.com/>
| FOLLOW
US <https://twitter.com/remindhq> | LIKE US
<https://www.facebook.com/remindhq>
