Segment hand-off not happening

I am new to druid and currently I am running a real-time node and it is injesting data and creating segments but it is not able to hand-off the segments to the historical node. Below are the spec file and the logs.
Can someone please validate and let me know if I am missing anything here.

[

{

“dataSchema” : {

“dataSource” : “ClickDataSource”,

“parser” : {

“type” : “string”,

“parseSpec” : {

“format” : “json”,

“timestampSpec” : {

“column” : “ClickDate”,

“format” : “auto”

},

“dimensionsSpec” : {

“dimensions”: [“ClickId”],

“dimensionExclusions” : ,

“spatialDimensions” :

}

}

},

“metricsSpec” : [{

“type” : “count”,

“name” : “count”

}],

“granularitySpec” : {

“type” : “uniform”,

“segmentGranularity” : “HOUR”,

“queryGranularity” : “NONE”

}

},

“ioConfig” : {

“type” : “realtime”,

“firehose” : {

“type” : “rabbitmq”,

“connection” : {

“host”: “localhost”,

“port”: “5672”,

“username”: “druid”,

“password”: “druid”

},

“config” : {

“exchange”: “druid”,

“queue” : “druidtest”,

“routingKey”: “#”,

“durable”: “true”,

“exclusive”: “false”,

“autoDelete”: “false”,

“maxRetries”: “10”,

“retryIntervalSeconds”: “1”,

“maxDurationSeconds”: “300”

}

},

“plumber”: {

“type”: “realtime”

}

},

“tuningConfig”: {

“type” : “realtime”,

“maxRowsInMemory”: 500000,

“intermediatePersistPeriod”: “PT10m”,

“windowPeriod”: “PT10m”,

“basePersistDirectory”: “/tmp/realtime/basePersist”,

“rejectionPolicy”: {

“type”: “windowTime”

}

}

}

]

2015-12-14T14:51:14,769 INFO [chief-ClickDataSource[0]] io.druid.server.coordination.BatchDataSegmentAnnouncer - Announcing segment[ClickDataSource_2015-12-14T10:00:00.000-08:00_2015-12-14T11:00:00.000-08:00_2015-12-14T10:00:00.000-08:00] at path[/druid/segments/localhost:8084/localhost:8084_realtime__default_tier_2015-12-14T14:51:14.470-08:00_35d0b6c91de248dbb47707f7fa09c9310]

2015-12-14T14:51:14,934 INFO [chief-ClickDataSource[0]] io.druid.segment.realtime.plumber.RealtimePlumber - Expect to run at [2015-12-14T15:10:00.000-08:00]

2015-12-14T14:51:14,941 INFO [chief-ClickDataSource[0]] io.druid.segment.realtime.plumber.RealtimePlumber - Starting merge and push.

2015-12-14T14:51:14,942 INFO [chief-ClickDataSource[0]] io.druid.segment.realtime.plumber.RealtimePlumber - Found [5] segments. Attempting to hand off segments that start before [1969-12-31T16:00:00.000-08:00].

2015-12-14T14:51:14,942 INFO [chief-ClickDataSource[0]] io.druid.segment.realtime.plumber.RealtimePlumber - Skipping persist and merge for entry [1450126800000=Sink{interval=2015-12-14T13:00:00.000-08:00/2015-12-14T14:00:00.000-08:00, schema=DataSchema{dataSource=‘ClickDataSource’, parser={type=string, parseSpec={format=json, timestampSpec={column=ClickDate, format=auto}, dimensionsSpec={dimensions=[ClickId, ClickType, ParentClick, ChildClick], dimensionExclusions=, spatialDimensions=}}}, aggregators=[CountAggregatorFactory{name=‘count’}], granularitySpec=io.druid.segment.indexing.granularity.UniformGranularitySpec@5bd85251}}] : Start time [2015-12-14T13:00:00.000-08:00] >= [1969-12-31T16:00:00.000-08:00] min timestamp required in this run. Segment will be picked up in a future run.

2015-12-14T14:51:14,942 INFO [chief-ClickDataSource[0]] io.druid.segment.realtime.plumber.RealtimePlumber - Skipping persist and merge for entry [1450130400000=Sink{interval=2015-12-14T14:00:00.000-08:00/2015-12-14T15:00:00.000-08:00, schema=DataSchema{dataSource=‘ClickDataSource’, parser={type=string, parseSpec={format=json, timestampSpec={column=ClickDate, format=auto}, dimensionsSpec={dimensions=[ClickId, ClickType, ParentClick, ChildClick], dimensionExclusions=, spatialDimensions=}}}, aggregators=[CountAggregatorFactory{name=‘count’}], granularitySpec=io.druid.segment.indexing.granularity.UniformGranularitySpec@5bd85251}}] : Start time [2015-12-14T14:00:00.000-08:00] >= [1969-12-31T16:00:00.000-08:00] min timestamp required in this run. Segment will be picked up in a future run.

2015-12-14T14:51:14,943 INFO [chief-ClickDataSource[0]] io.druid.segment.realtime.plumber.RealtimePlumber - Skipping persist and merge for entry [1450123200000=Sink{interval=2015-12-14T12:00:00.000-08:00/2015-12-14T13:00:00.000-08:00, schema=DataSchema{dataSource=‘ClickDataSource’, parser={type=string, parseSpec={format=json, timestampSpec={column=ClickDate, format=auto}, dimensionsSpec={dimensions=[ClickId, ClickType, ParentClick, ChildClick], dimensionExclusions=, spatialDimensions=}}}, aggregators=[CountAggregatorFactory{name=‘count’}], granularitySpec=io.druid.segment.indexing.granularity.UniformGranularitySpec@5bd85251}}] : Start time [2015-12-14T12:00:00.000-08:00] >= [1969-12-31T16:00:00.000-08:00] min timestamp required in this run. Segment will be picked up in a future run.

2015-12-14T14:51:14,943 INFO [chief-ClickDataSource[0]] io.druid.segment.realtime.plumber.RealtimePlumber - Skipping persist and merge for entry [1450119600000=Sink{interval=2015-12-14T11:00:00.000-08:00/2015-12-14T12:00:00.000-08:00, schema=DataSchema{dataSource=‘ClickDataSource’, parser={type=string, parseSpec={format=json, timestampSpec={column=ClickDate, format=auto}, dimensionsSpec={dimensions=[ClickId, ClickType, ParentClick, ChildClick], dimensionExclusions=, spatialDimensions=}}}, aggregators=[CountAggregatorFactory{name=‘count’}], granularitySpec=io.druid.segment.indexing.granularity.UniformGranularitySpec@5bd85251}}] : Start time [2015-12-14T11:00:00.000-08:00] >= [1969-12-31T16:00:00.000-08:00] min timestamp required in this run. Segment will be picked up in a future run.

2015-12-14T14:51:14,943 INFO [chief-ClickDataSource[0]] io.druid.segment.realtime.plumber.RealtimePlumber - Skipping persist and merge for entry [1450116000000=Sink{interval=2015-12-14T10:00:00.000-08:00/2015-12-14T11:00:00.000-08:00, schema=DataSchema{dataSource=‘ClickDataSource’, parser={type=string, parseSpec={format=json, timestampSpec={column=ClickDate, format=auto}, dimensionsSpec={dimensions=[ClickId, ClickType, ParentClick, ChildClick], dimensionExclusions=, spatialDimensions=}}}, aggregators=[CountAggregatorFactory{name=‘count’}], granularitySpec=io.druid.segment.indexing.granularity.UniformGranularitySpec@5bd85251}}] : Start time [2015-12-14T10:00:00.000-08:00] >= [1969-12-31T16:00:00.000-08:00] min timestamp required in this run. Segment will be picked up in a future run.

2015-12-14T14:51:14,943 INFO [chief-ClickDataSource[0]] io.druid.segment.realtime.plumber.RealtimePlumber - Found [0] sinks to persist and merge

Correct Spec

[

{

“dataSchema” : {

“dataSource” : “ClickDataSource”,

“parser” : {

“type” : “string”,

“parseSpec” : {

“format” : “json”,

“timestampSpec” : {

“column” : “ClickDate”,

“format” : “auto”

},

“dimensionsSpec” : {

“dimensions”: [“ClickId”],

“dimensionExclusions” : ,

“spatialDimensions” :

}

}

},

“metricsSpec” : [{

“type” : “count”,

“name” : “count”

}],

“granularitySpec” : {

“type” : “uniform”,

“segmentGranularity” : “HOUR”,

“queryGranularity” : “NONE”

}

},

“ioConfig” : {

“type” : “realtime”,

“firehose” : {

“type” : “rabbitmq”,

“connection” : {

“host”: “localhost”,

“port”: “5672”,

“username”: “druid”,

“password”: “druid”

},

“config” : {

“exchange”: “druid”,

“queue” : “druidtest”,

“routingKey”: “#”,

“durable”: “true”,

“exclusive”: “false”,

“autoDelete”: “false”,

“maxRetries”: “10”,

“retryIntervalSeconds”: “1”,

“maxDurationSeconds”: “300”

}

},

“plumber”: {

“type”: “realtime”

}

},

“tuningConfig”: {

“type” : “realtime”,

“maxRowsInMemory”: 500000,

“intermediatePersistPeriod”: “PT10m”,

“windowPeriod”: “PT10m”,

“basePersistDirectory”: “/tmp/realtime/basePersist”,

“rejectionPolicy”: {

“type”: “serverTime”

}

}

}

]

Hey Raguvind,

I don’t see any obvious issues with your spec file. The realtime node will only handoff segments to the historical nodes every “segmentGranularity” + “windowPeriod”, so in your case 1 hour + 10 minutes. For development purposes, you may want to increase the handoff frequency. You can set “segmentGranularity” to “MINUTE” and “windowPeriod” to “PT10s” to have it handoff every 70 seconds.

Make sure that you are indeed using “serverTime” as the rejection policy and then wait for 70 seconds. Seeing ‘Attempting to hand off segments that start before [1969-12-31T16:00:00.000-08:
00]’ makes me suspect that the rejection policy might be set to none or messageTime.

Please see: http://druid.io/docs/latest/ingestion/faq.html