input {
	kafka {
		bootstrap_servers => "10.0.1.10:9092"
		group_id => "cheonan"
		topics => ["cheonan-kt"]
		codec => json {
			charset=>"UTF-8"
		}
	}
}

# filter {
# 	mutate {
# 		split => {"message" => " "}
# 	}
# }

output {
#	stdout {
#		codec => rubydebug
#	}
#	stdout { codec => line { format => 'tttttttttttttttttttttttt' } }
	if [data_type] == "cheonan_kt" {
		jdbc {
			connection_string => "jdbc:postgresql://pg-219s5.vpc-cdb-kr.gov-ntruss.com:5432/cheonan_ep?user=cheonan&password=cjsdkstl2023!#"
			driver_jar_path => "/usr/share/logstash/postgresql-42.2.24.jar"
			max_pool_size => 2
			connection_timeout => 25000
			statement => [
				"INSERT INTO energy.raw_data_kt (
					tm,
					request_type,
					client_host,
					body
				) VALUES (
					now(),
					?,
					?,
					?
				)",
				"[cheonan_kt][request_type]",
				"[cheonan_kt][client_host]",
				"[cheonan_kt][body]"
			]
		}
	}
}