Duplicate record to each index while parsing JSON by individual TCP Port

Hi all, i have meet a very tricky case while parsing data by individual TCP port.

To reproduce the case, it is very simple,
#

For all general issues, please provide the following details for fast resolution:
- Version:
/opt/elasticsearch/bin/elasticsearch --version
Version: 2.4.0, Build: ce9f0c7/2016-08-29T09:14:17Z, JVM: 1.8.0_92
/opt/logstash/bin/logstash --version
logstash 2.3.4
- Operating System:
CentOS 6.4
- Config File (if you have sensitive info, please remove it):
- Sample Data:
Under /etc/logstash/conf.d/
create sample1.conf & sample2.conf & sample3.conf

Configuration as follows.
sample1.conf===============================
input {
tcp {
port => 50001
type => "sample1"
codec => "json"
}
}

filter {
if [type] == "sample1" {
grok {
remove_field => target
remove_field => "tags"
}
}
}

output {
elasticsearch {
hosts => ["es-db01.com:9200"]
index => "sample1-%{+YYYY.MM.dd}"
workers => 1
}
stdout { codec => rubydebug }
}

sample2.conf===============================
input {
tcp {
port => 50002
type => "sample2"
codec => "json"
}
}

filter {
if [type] == "sample2" {
grok {
remove_field => target
remove_field => "tags"
}
}
}

output {
elasticsearch {
hosts => ["es-db01.com:9200"]
index => "sample2-%{+YYYY.MM.dd}"
workers => 1
}
stdout { codec => rubydebug }
}

sample3.conf===============================
input {
tcp {
port => 50003
type => "sample3"
codec => "json"
}
}

filter {
if [type] == "sample3" {
grok {
remove_field => target
remove_field => "tags"
}
}
}

output {
elasticsearch {
hosts => ["es-db01.com:9200"]
index => "sample3-%{+YYYY.MM.dd}"
workers => 1
}
stdout { codec => rubydebug }
}
- Steps to Reproduce:

After setup logstash configuration, we restart by "service logstash restart", then all port is LISTEN right now.

I am writing ruby script to parse the data.

Ruby Code Start-------------------
require 'logstash-logger'
@ls_logger = LogStashLogger.new(type: :tcp, host: 'es-db01.com', port: 50001)

def parse_to_elk(data)
@ls_logger.info data.to_json
end

test_hash = {}
test_hash[:hash_a] = "test_a"
test_hash[:hash_b] = "test_b"

p test_hash
parse_to_elk(test_hash)

Ruby Code End--------------------

Then, 3 duplicated records will be inserted in Index "sample1-2016XX", "sample2-2016XX", "sample3-2016XX"

Suppose the 1 record will be inserted to sample1 only......but i have no idea other index also have the same data..........