How to migrate parent child relation data from RDBMS, through logstash to Elastic?

[prasad1986 (prasad)]

I created Parent and child relation through Kibana tool.
Index :demosampleindex
documenttype: itemdoctype
Parent Relation:itemjoin
child Relation :itemjoin

Syntax:
PUT demosampleindex
{
"settings": {
"number_of_shards": 1,
"number_of_replicas": 1
},
"mappings": {
"itemdoctype": {
"properties": {
"itemjoinfield": {
"type": "join",
"relations": {
"itemjoin": "cfjoin"
}
}
}
}
}
}

After creation of index try to move data from logstash. Parent reation data able migrate it, but child relation data not able to achive it.
"item_id" is document_id of parent object.

Logstash Syntax:
input {
jdbc {

    jdbc_connection_string => "jdbc:oracle:thin:@192.168.142.157:1521:nios157"
    	
    jdbc_user => "C2_DEV_NIOS5_OLTP2"
  		
	jdbc_password => "nios"
	 
    jdbc_driver_library => "D:/SVN/PRODUCT/NIOS/Engg/Dev/6.0/lib/ojdbc6.jar"
    					
    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
    				
	statement_filepath => "select distinct  item.item_id  as  itemid, item.item_id||'_cfobj' as itemidstr,'cfjoin' as cfjoinname, item.customfieldname  as customfieldname, item.customfieldvalue as customfieldvalue from  item_table item "

}

}
filter {

mutate {
    add_field => { "_parent" => "%{itemid}" }
}

 aggregate {
   task_id => "%{itemid}"
   code => "
    map['itemidstr'] ||= event.get('itemidstr')

	map['itemjoinfield'] = {
		'name' => event.get('cfjoinname')
	}						
						
	map['cf_list'] ||= []
	map['customfields'] ||= []
	if (event.get('customfieldname') != nil)
		if !( map['cf_list'].include? event.get('customfieldname') ) 
		 map['cf_list'] << event.get('customfieldname')
			  map['customfields'] << {
				'customfieldname' => event.get('customfieldname'),
				'customfieldvalue' => event.get('customfieldvalue')
			}
		end	
	end
	
	"

   push_previous_map_as_event => true
   timeout => 150000
 }
mutate { 
	remove_field => ["cf_list"]
}

}
output {
elasticsearch {
index => "demosampleindex"
document_type => "itemdoctype"
document_id => "%{itemidstr}"
hosts => ["localhost:9200"]
routing => "%{itemid}"
parent => "%{itemid}"
}
}

Exception log:

[2019-02-25T16:21:08,163][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>"576044_cfobj", :_index=>"demosampleindex", :_type=>"itemdoctype", :routing=>"576044", :parent=>"576044"}, #LogStash::Event:0x48d2b7f4], :response=>{"index"=>{"_index"=>"demosampleindex", "_type"=>"itemdoctype", "_id"=>"576044_cfobj", "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"can't specify parent if no parent field has been configured"}}}}
[2019-02-25T16:21:08,214][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>"576044_cfobj", :_index=>"demosampleindex", :_type=>"itemdoctype", :routing=>"576044", :parent=>"576044"}, #LogStash::Event:0x30a471ba], :response=>{"index"=>{"_index"=>"demosampleindex", "_type"=>"itemdoctype", "_id"=>"576044_cfobj", "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"can't specify parent if no parent field has been configured"}}}}
[2019-02-25T16:21:08,216][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>"576044_cfobj", :_index=>"demosampleindex", :_type=>"itemdoctype", :routing=>"576044", :parent=>"576044"}, #LogStash::Event:0x35eeb606], :response=>{"index"=>{"_index"=>"demosampleindex", "_type"=>"itemdoctype", "_id"=>"576044_cfobj", "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"can't specify parent if no parent field has been configured"}}}}
[2019-02-25T16:21:08,294][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>"576044_cfobj", :_index=>"demosampleindex", :_type=>"itemdoctype", :routing=>"%{itemid}", :parent=>"%{itemid}"}, #LogStash::Event:0x5f3f7658], :response=>{"index"=>{"_index"=>"demosampleindex", "_type"=>"itemdoctype", "_id"=>"576044_cfobj", "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"can't specify parent if no parent field has been configured"}}}}
[2019-02-25T16:21:08,298][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>"576044_cfobj", :_index=>"demosampleindex", :_type=>"itemdoctype", :routing=>"576044", :parent=>"576044"}, #LogStash::Event:0x2ea1e71f], :response=>{"index"=>{"_index"=>"demosampleindex", "_type"=>"itemdoctype", "_id"=>"576044_cfobj", "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"can't specify parent if no parent field has been configured"}}}}
[2019-02-25T16:21:08,300][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>"576044_cfobj", :_index=>"demosampleindex", :_type=>"itemdoctype", :routing=>"576044", :parent=>"576044"}, #LogStash::Event:0x3875c020], :response=>{"index"=>{"_index"=>"demosampleindex", "_type"=>"itemdoctype", "_id"=>"576044_cfobj", "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"can't specify parent if no parent field has been configured"}}}}
[2019-02-25T16:21:08,726][INFO ][logstash.pipeline ] Pipeline has terminated {:pipeline_id=>"main", :thread=>"#<Thread:0x4562c90c run>"}