Slow response with jdbc postgresql sink

[xiangjinwu]

Reproduce:

• Have a PostgreSQL server running.
• Create a table as sink in PostgreSQL: create table sk1 (v int primary key);
• ./risedev configure to enable building connector node
• Edit risedev.yml to uncomment - use: connector-node
• ./risedev p to start RisingWave in playground mode

create table t (v int primary key);
create materialized view mv as select v from t;
insert into t values (32); -- returns instantly

create sink sk from mv with (connector='jdbc', jdbc.url='jdbc:postgresql://127.0.0.1:5432/test?username=USERNAME', table.name='sk1', type='upsert', primary_key='v');
insert into t values (41); -- slow

Note that both rw_implicit_flush and sink_decouple default to false.

• If a flush; command is issued or rw_implicit_flush is true, the current session will stuck and cannot be canceled with Ctrl-C.
• If sink_decouple is true, risingwave-streaming-actor would panic, with the connector java process still running and listening on port 50051.

[StrikeW]

I reproduced the issue on my local env (a7b15d8). By default, the playground will utilize all cores of the machine (10 in my local env), thus the parallelism of mv and sink are both 10.
I found that the issue can also reproduce in the append-only sink, so it has nothing to do with the additional shuffle for upsert sink (#11042). When we drop the sink sk and recreate it with set streaming_parallelism=1, the INSERT command can return instantly.
I checked the async stack trace, it shows the sink executors cost the most of the time. Still analyzing the issue.
Btw, @wenym1 Do you have any ideas?

>> Actor 31
Actor 31: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.292s]
  Epoch 4859972811751424 [8.642s]
    SinkExecutor 1F00000002 (actor 31, operator 2) [8.642s]  <== current

>> Actor 32
Actor 32: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.292s]
  Epoch 4859972811751424 [4.802s]
    SinkExecutor 2000000002 (actor 32, operator 2) [4.802s]  <== current

>> Actor 33
Actor 33: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.292s]
  Epoch 4859972811751424 [6.734s]
    SinkExecutor 2100000002 (actor 33, operator 2) [6.734s]  <== current

>> Actor 34
Actor 34: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.292s]
  Epoch 4859972811751424 [5.776s]
    SinkExecutor 2200000002 (actor 34, operator 2) [5.776s]  <== current

>> Actor 35
Actor 35: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.292s]
  Epoch 4859972811751424 [7.692s]
    SinkExecutor 2300000002 (actor 35, operator 2) [7.692s]  <== current

>> Actor 36
Actor 36: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.292s]
  Epoch 4859972811751424 [3.852s]
    SinkExecutor 2400000002 (actor 36, operator 2) [3.852s]  <== current

>> Actor 37
Actor 37: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.291s]
  Epoch 4859972811751424 [2.878s]
    SinkExecutor 2500000002 (actor 37, operator 2) [2.878s]  <== current

>> Actor 38
Actor 38: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.291s]
  Epoch 4859972811751424 [897.625ms]
    SinkExecutor 2600000002 (actor 38, operator 2) [897.601ms]  <== current

>> Actor 39
Actor 39: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.291s]
  Epoch 4859972811751424 [1.900s]
    SinkExecutor 2700000002 (actor 39, operator 2) [1.900s]  <== current

>> Actor 40
Actor 40: `CREATE SINK sk1 FROM mv WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://127.0.0.1:8432/mydb?user=myuser&password=123456', table.name = 'sk1', type = 'upsert', primary_key = 'v')` [108.291s]
  Epoch 4859972746280960 [!!! 19.533s]
    SinkExecutor 2800000002 (actor 40, operator 2) [!!! 19.533s]  <== current

[wenym1]

The panic of sink decouple is a known issue and should be fixed merge #11017 is merged.

[wenym1]

I did some investigation. I noticed that the checkpoint interval is 1 second, and a commit round trip between CN and connector node sometimes takes more than 1 second, which makes our barrier piles up. This might be the reason for slow response. The newly issued flush should wait for all previous barriers to be processed.

[wenym1]

Fixed with #11589