[jira] [Created] (FLUME-3138) SchemaURL from flume configuration is dropping the flume events expecting the schema url to be added in event header as against FLUME-2810

["Ramgopal N (JIRA)" <ji...@apache.org>]

Ramgopal N created FLUME-3138:
---------------------------------

             Summary: SchemaURL from flume configuration is dropping the flume events expecting the schema url to be added in event header as against FLUME-2810
                 Key: FLUME-3138
                 URL: https://issues.apache.org/jira/browse/FLUME-3138
             Project: Flume
          Issue Type: Question
          Components: Configuration
    Affects Versions: 1.7.0
         Environment: Flume1.7

            Reporter: Ramgopal N


I have avro data coming to kafka topic. Flume reads the events from kafka and then using kite dataset with hdfs sink is put into HDFS as parquet data.
Flume config is as below:
agent.sinks.k1.channel = c1
agent.sinks.k1.type = org.apache.flume.sink.kite.DatasetSink
agent.sinks.k1.kite.dataset.uri = dataset:hdfs://namenodeHA/kite/avro_to_parquet_item2
agent.sinks.k1.serializer = org.apache.flume.sink.hdfs.AvroEventSerializer$Builder
agent.sinks.k1.hdfs.filePrefix=parquetdata
agent.sinks.k1.hdfs.fileSuffix = .parquet
agent.sinks.k1.hdfs.fileType=DataStream
#agent.sinks.k1.hdfs.rollInterval=30
#agent.sinks.k1.hdfs.rollCount=1
#agent.sinks.k1.hdfs.batchSize=1
agent.sinks.k1.kite.batchSize=2
agent.sinks.k1.kite.rollInterval=30
agent.sinks.k1.kite.flushable.commitOnBatch=true
#agent.sinks.k1.hdfs.path = hdfs://namenodeHA/user/flumetest
#agent.sinks.k1.serializer.compressionCodec = snappy
agent.sinks.k1.serializer.schemaURL = hdfs://namenodeHA/kite/item.avsc

I am getting the below exception in the flume logs:
2017-07-31 06:18:40,796 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.hadoop.io.compress.CodecPool.getCompressor(CodecPool.java:153)] Got brand-new compressor [.snappy]
2017-07-31 06:18:40,802 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.kitesdk.data.spi.filesystem.FileSystemWriter.initialize(FileSystemWriter.java:147)] Opened output appender ParquetAppender{path=hdfs://namenodeHA/kite/avro_to_parquet_item2/.6d1019b3-96c4-4334-b737-af260d17aac4.parquet.tmp, schema={"type":"record","name":"item","namespace":"item.avro","fields":[{"name":"i_item_sk","type":..................................{"name":"i_manager_id","type":["null","int"]},{"name":"i_product_name","type":["null","string"]}]}, fileSystem=DFS[DFSClient[clientName=DFSClient_NONMAPREDUCE_2077692400_17, ugi=root (auth:SIMPLE)]], avroParquetWriter=parquet.avro.AvroParquetWriter@31ffba30} for hdfs://namenodeHA/kite/avro_to_parquet_item2/6d1019b3-96c4-4334-b737-af260d17aac4.parquet
2017-07-31 06:18:40,803 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - org.apache.flume.sink.kite.policy.RetryPolicy.handle(RetryPolicy.java:39)] Event delivery failed: No schema in event headers. Headers must include either flume.avro.schema.url or flume.avro.schema.literal
2017-07-31 06:18:40,803 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:158)] Unable to deliver event. Exception follows.
org.apache.flume.EventDeliveryException: org.apache.flume.sink.kite.NonRecoverableEventException: No schema in event headers. Headers must include either flume.avro.schema.url or flume.avro.schema.literal
	at org.apache.flume.sink.kite.policy.RetryPolicy.handle(RetryPolicy.java:42)
	at org.apache.flume.sink.kite.DatasetSink.write(DatasetSink.java:375)
	at org.apache.flume.sink.kite.DatasetSink.process(DatasetSink.java:301)
	at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67)
	at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145)
	at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.flume.sink.kite.NonRecoverableEventException: No schema in event headers. Headers must include either flume.avro.schema.url or flume.avro.schema.literal
	at org.apache.flume.sink.kite.parser.AvroParser.schema(AvroParser.java:185)
	at org.apache.flume.sink.kite.parser.AvroParser.parse(AvroParser.java:155)
	at org.apache.flume.sink.kite.parser.AvroParser.parse(AvroParser.java:56)
	at org.apache.flume.sink.kite.DatasetSink.write(DatasetSink.java:366)




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)