Hi Muhammad,
File channel converts the events to its internal FlumeEvent class (
https://github.com/apache/flume/blob/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FlumeEvent.java)
during the put operation:
https://github.com/apache/flume/blob/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java#L649-L650
When the sink takes the events from the channel it gets FlumeEvent
instances (in case of file channel), that is the reason of the exception
you got.
Note that the file channel only stores the event's body and headers, no
other extra properties will be persisted thus returned on take operations.
I don't know the exact details of your event implementation, but I'd
recommend to instead of casting the event to ServerEvent create a
constructor in ServerEvent which gets an org.apache.flume.Event (or byte[]
body, Map<String, String> headers parameters) and uses this to construct
itself.
Regards,
Denes
On Thu, Aug 24, 2017 at 2:33 PM Muhammad Yaseen <yaseenaftab...@gmail.com>
wrote:
> Hello,
>
> I have implemented a custom HTTPSource handler, and am not using the
> default { "headers": { } , "body": "text" } format. Hence, I am also not
> using the default JSONEvent class and have defined my own class (which also
> implements Event interface).
>
> class ServerEvent implements Event.
>
> It has 3 additional attributes (String server,String output,long
> timestamp) which are mapped to the json object keys by Gson in the my
> handler class (ServerEventHandler) .
>
> The format of JSON post data received: { "server": "some server",
> "output": "some output", "timestamp": 1234567890 }
>
> I have also implemented a custom Avro Serializer for this event class (by
> extending from AbstractAvroEventSerializer<T>).
>
> In the convert(Event e) method I am doing the following.
> {
> ServerEvent se = (ServerEvent)event; <-- exception on this line
> // do something with 'se'
> }
>
> This works fine when I use the Memory Channel, but throws the following
> exception if I use the File Channel:
>
> 7/08/24 07:33:16 ERROR flume.SinkRunner: Unable to deliver event.
> Exception follows.
> org.apache.flume.EventDeliveryException: java.lang.ClassCastException:
> org.apache.flume.channel.file.FlumeEvent cannot be cast to
> com.myapp.flumeapp.events.ServerEvent
> at org.apache.flume.sink.hdfs.HDFSEventSink.process(HDFSEventSink.java:451)
> 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:748)
> Caused by: java.lang.ClassCastException:
> org.apache.flume.channel.file.FlumeEvent cannot be cast to
> com.myapp.flumeapp.events.ServerEvent
> at com.myapp.flumeapp
> .eventserializers.ServerEventAvroSerializer.convert(ServerEventAvroSerializer.java:46)
> at com.myapp.flumeapp
> .eventserializers.ServerEventAvroSerializer.convert(ServerEventAvroSerializer.java:15)
> at
> org.apache.flume.serialization.AbstractAvroEventSerializer.write(AbstractAvroEventSerializer.java:107)
> at
> org.apache.flume.sink.hdfs.HDFSDataStream.append(HDFSDataStream.java:119)
> at org.apache.flume.sink.hdfs.BucketWriter$7.call(BucketWriter.java:540)
> at org.apache.flume.sink.hdfs.BucketWriter$7.call(BucketWriter.java:537)
> at org.apache.flume.sink.hdfs.BucketWriter$9$1.run(BucketWriter.java:668)
> at
> org.apache.flume.auth.SimpleAuthenticator.execute(SimpleAuthenticator.java:50)
> at org.apache.flume.sink.hdfs.BucketWriter$9.call(BucketWriter.java:665)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
>
> The same issue is mentioned here:
> https://github.com/telefonicaid/fiware-cygnus/issues/1419
>
> They trace the issue to custom event class not implementing the Writable
> interface.
>
> Has anyone else tried custom HTTP handler and serializer with file
> channel? Any way to get around this limitation ?
>
> --
>
> Regards,
> Muhammad Yaseen
>
