Rsyslog, Logstash parse Error when incoming log is too big

Hello, I have build a syslog server with web gui for this is use the ELK-Stack and configured Rsyslog. Everything is good and it works but when clients send log files that are more than 7999 characters long, I get the following error message:

@timestamp:Oct 12, 2021 @ 15:27:24.846 @version:1 host:localhost message:{"@timestamp":"2021-10-12T15:27:24.801983+02:00","@version":"1","message":" [2021-10-12T15:27:24,801][WARN ][logstash.codecs.jsonlines][main][125ec643269767cb0518268efbd2d657c1d9ad103ddcdfba87dbb3b48aa73999] JSON parse error, original data now in message field {:message=>"Unexpected end-of-input: was expecting closing quote for a string value\n at [Source: (String)\"{\"@timestamp\":\"2021-10-12T15:27:24.758140+02:00\",\"@version\":\"1\",\"message\":\" [2021-10-12T15:27:24,757][WARN ][logstash.codecs.jsonlines][main][125ec643269767cb0518268efbd2d657c1d9ad103ddcdfba87dbb3b48aa73999] JSON parse error, original data now in message field {:message=>\\\"Unexpected end-of-input: was expecting closing quote for a string value\\\\n at [Source: (String)\\\\\\\"

Big problem is that Logstash is spaming this message 10000 times in seconds and endless, It creates more than 10GB Logs in seconds. I can just stop it, when I restart logstash.

Elasticsearch.yml

# ======================== Elasticsearch Configuration =========================
#
# NOTE: Elasticsearch comes with reasonable defaults for most settings.
#       Before you set out to tweak and tune the configuration, make sure you
#       understand what are you trying to accomplish and the consequences.
#
# The primary way of configuring a node is via this file. This template lists
# the most important settings you may want to configure for a production cluster.
#
# Please consult the documentation for further information on configuration options:
# https://www.elastic.co/guide/en/elasticsearch/reference/index.html
#
# ---------------------------------- Cluster -----------------------------------
#
# Use a descriptive name for your cluster:
#
cluster.name: Syslog
#
# ------------------------------------ Node ------------------------------------
#
# Use a descriptive name for the node:
#
node.name: node-1
#
# Add custom attributes to the node:
#
#node.attr.rack: r1
#
# ----------------------------------- Paths ------------------------------------
#
# Path to directory where to store the data (separate multiple locations by comma):
#
path.data: /var/lib/elasticsearch
#
# Path to log files:
#
path.logs: /var/log/elasticsearch
#
# ----------------------------------- Memory -----------------------------------
#
# Lock the memory on startup:
#
#bootstrap.memory_lock: true
#
# Make sure that the heap size is set to about half the memory available
# on the system and that the owner of the process is allowed to use this
# limit.
#
# Elasticsearch performs poorly when the system is swapping the memory.
#
# ---------------------------------- Network -----------------------------------
#
# By default Elasticsearch is only accessible on localhost. Set a different
# address here to expose this node on the network:
#
network.host: ["10.0.15.44", "localhost"]
#
# By default Elasticsearch listens for HTTP traffic on the first free port it
# finds starting at 9200. Set a specific HTTP port here:
#
#http.port: 9200
#
# For more information, consult the network module documentation.
#
# --------------------------------- Discovery ----------------------------------
#
# Pass an initial list of hosts to perform discovery when this node is started:
# The default list of hosts is ["127.0.0.1", "[::1]"]
#
discovery.seed_hosts: ["10.0.15.44", "localhost"]
#
# Bootstrap the cluster using an initial set of master-eligible nodes:
#
#cluster.initial_master_nodes: ["node-1", "node-2"]
#
# For more information, consult the discovery and cluster formation module documentation.
#
# ---------------------------------- Various -----------------------------------
#
# Require explicit names when deleting indices:
#
#action.destructive_requires_name: true
#
# ---------------------------------- Various ----------------------------------
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: elastic-certificates.p12

xpack.security.http.ssl.enabled: true
xpack.security.http.ssl.keystore.path: elastic-certificates.p12
xpack.security.http.ssl.truststore.path: elastic-certificates.p12
xpack.security.http.ssl.client_authentication: optional

My Logstash Configuration Input / Output:

input {
  tcp {
    type => "rsyslog"
    host => "127.0.0.1"
    port => 10514
	codec => "json_lines"
  }
 }
 
filter {


}


# Every single log will be forwarded to ElasticSearch. If you are using another port, you should specify it here.
output {
  if [type] == "rsyslog" {
    elasticsearch {
      hosts => [ "https://localhost:9200" ]
	  user => elastic
	  password => XXX
	  cacert => "/etc/logstash/newfile.crt.pem"
      ssl_certificate_verification => false
    }
  }
}



My Rsyslog JSON Template:

template(name="ls_json" type="list" option.json="on") {
  constant(value="{")
  constant(value="\"@timestamp\":\"") property(name="timereported" dateFormat="rfc3339")
  constant(value="\",\"@version\":\"1")
  constant(value="\",\"message\":\"") property(name="msg")
  constant(value="\",\"host\":\"") property(name="hostname")
  constant(value="\",\"severity\":\"") property(name="syslogseverity-text")
  constant(value="\",\"facility\":\"") property(name="syslogfacility-text")
  constant(value="\",\"programname\":\"") property(name="programname")
  constant(value="\",\"procid\":\"") property(name="procid")
  constant(value="\"}\n")
} 

Rsyslog Output Conf forward to Logstash:

*.*  @@127.0.0.1:10514;ls_json

Rsyslog.conf

# /etc/rsyslog.conf configuration file for rsyslog
#
# For more information install rsyslog-doc and see
# /usr/share/doc/rsyslog-doc/html/configuration/index.html


#################
#### MODULES ####
#################

module(load="imuxsock") # provides support for local system logging
module(load="imklog")   # provides kernel logging support
#module(load="immark")  # provides --MARK-- message capability

# provides UDP syslog reception
module(load="imudp")
input(type="imudp" port="514")

# provides TCP syslog reception
module(load="imtcp")
input(type="imtcp" port="514")


###########################
#### GLOBAL DIRECTIVES ####
###########################

#
# Use traditional timestamp format.
# To enable high precision timestamps, comment out the following line.
#
$ActionFileDefaultTemplate RSYSLOG_TraditionalFileFormat

#
# Set the default permissions for all log files.
#
$FileOwner root
$FileGroup adm
$FileCreateMode 0640
$DirCreateMode 0755
$Umask 0022

#
# Where to place spool and state files
#
$WorkDirectory /var/spool/rsyslog

#
# Include all config files in /etc/rsyslog.d/
#
$IncludeConfig /etc/rsyslog.d/*.conf


#
# Debugging Rsyslog 
#
$DebugFile /var/log/rsyslog-debug.log
$DebugLevel 2


###############
#### RULES ####
###############

#
# First some standard log files.  Log by facility.
#
auth,authpriv.*			/var/log/auth.log
*.*;auth,authpriv.none		-/var/log/syslog
#cron.*				/var/log/cron.log
daemon.*			-/var/log/daemon.log
kern.*				-/var/log/kern.log
lpr.*				-/var/log/lpr.log
mail.*				-/var/log/mail.log
user.*				-/var/log/user.log

#
# Logging for the mail system.  Split it up so that
# it is easy to write scripts to parse these files.
#
mail.info			-/var/log/mail.info
mail.warn			-/var/log/mail.warn
mail.err			/var/log/mail.err

#
# Some "catch-all" log files.
#
*.=debug;\
	auth,authpriv.none;\
	news.none;mail.none	-/var/log/debug
*.=info;*.=notice;*.=warn;\
	auth,authpriv.none;\
	cron,daemon.none;\
	mail,news.none		-/var/log/messages

#
# Emergencies are sent to everybody logged in.
#
*.emerg				:omusrmsg:*