mysql connection unexpected

[ctbsea]

[mysql] 2023/08/16 09:47:33 packets.go:73: unexpected EOF
[mysql] 2023/08/16 09:47:33 packets.go:428: busy buffer
{"@service":"benthos","label":"f0a6ffcfac2d8670225c7e92e8e4f553de3f505b3526d087db4bcda6c43fd293_20230816174650","level":"error","msg":"Failed to read message: invalid connection","path":"root.input","stream":"cfc020a2-082e-4e5b-a713-049a9678ead8"}

{
    "active":false,
    "uptime":42.554192632,
    "uptime_str":"42.554192632s",
    "config":{
        "buffer":{
            "none":{

            }
        },
        "input":{
            "label":"f0a6ffcfac2d8670225c7e92e8e4f553de3f505b3526d087db4bcda6c43fd293_20230816174650",
            "sql_raw":{
                "args_mapping":"[  ]",
                "conn_max_life_time":"10s",
                "driver":"mysql",
                "dsn":"**********/?charset=utf8mb4&timeout=5s&readTimeout=3s",
                "query":"SELECT `id`,`user_name`,`id_card` FROM `TEST`.`t_1` "
            }
        },
        "output":{
            "label":"f0a6ffcfac2d8670225c7e92e8e4f553de3f505b3526d087db4bcda6c43fd293_20230816174650_output",
            "mongodb":{
                "backoff":{
                    "initial_interval":"1s",
                    "max_elapsed_time":"30s",
                    "max_interval":"5s"
                },
                "batching":{
                    "byte_size":0,
                    "check":"",
                    "count":0,
                    "period":"",
                    "processors":[

                    ]
                },
                "collection":"temp_cfc020a2-082e-4e5b-a713-049a9678ead8",
                "database":"LABEL",
                "document_map":"root.node_1 = this.node_1\nroot.node_0 = this.node_0",
                "filter_map":"",
                "hint_map":"",
                "max_in_flight":64,
                "max_retries":3,
                "operation":"insert-one",
                "password":"",
                "upsert":false,
                "url":"mongodb://xxx:27017/?authSource=admin",
                "username":"",
                "write_concern":{
                    "j":false,
                    "w":"1",
                    "w_timeout":""
                }
            }
        },
        "pipeline":{
            "processors":[
                {
                    "label":"",
                    "mapping":"root.node_1 = this"
                },
                {
                    "branch":{
                        "processors":[
                            {
                                "label":"",
                                "sql_raw":{
                                    "args_mapping":"[ this.node_1.id ]",
                                    "conn_max_life_time":"10s",
                                    "driver":"mysql",
                                    "dsn":"*********8/?charset=utf8mb4&timeout=5s&readTimeout=3s",
                                    "query":"SELECT `id`,`user_id`,`integral`  FROM `TEST`.`t_2` WHERE  user_id = ?"
                                }
                            }
                        ],
                        "request_map":"",
                        "result_map":"root.node_0 = this"
                    },
                    "label":""
                },
                {
                    "label":"",
                    "mapping":"root = this.explode(\"node_0\")"
                },
                {
                    "label":"",
                    "unarchive":{
                        "format":"json_array"
                    }
                }
            ],
            "threads":2
        }
    }
}

The input table TEST.t_1 " has 1 million records, and the error mentioned above occurs when running a few thousand records at a time. How should I handle this?

[Jeffail]

Hey @ctbsea, sounds like the connection is getting lost before all rows could be consumed, you'll need to either break the queries down with some sort of pagination or find another approach.

Converting to a discussion as per: #2026