You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@camel.apache.org by Sherwin Pinto <sh...@narvar.com> on 2016/07/14 20:18:35 UTC
Route setup to upload file to s3 and put message on kafka queue
HI All,
My use case is as follows
Use case 1:
1. Pole ftp server for file.
2. Upload file to S3
3. Put message on to kafka with file meta info i. e filename, S3 bucket etc. This information is then used by a separate java process to download file and process it.
I have a working prototype but want to make sure I am doing this correctly. Here is the skeleton of the route config.
from(config.getCamelFTPConsumerString()).routeId(config.getRouteId())
.idempotentConsumer(ExpressionBuilder.headerExpression(Exchange.FILE_NAME),
JpaMessageIdRepository.jpaMessageIdRepository(postgresEntityManager, "FLAT_FILE_PROCESSOR"))
.eager(true).process(myProcessor)
.to(s3Config.getCamelS3Configuration())
.process(kafkaProducerMessageProcessor)
.to(kafkaConfig.getCamelKafkaConfig())
.log("EventProc - Uploaded file ${file:name} to S3”);
I am using a pipeline, the file is first uploaded to S3, then kafkaProducerMessageProcessor modifies the message and replaces the output with the a JSON string that represents the kafka message which is then routed to kafka
public void process(Exchange exchange) throws Exception {
String s3fileName=exchange.getIn().getHeader(Exchange.FILE_NAME_PRODUCED).toString();
exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, null);
QueueMessage<S3FileLink> queueMsg= S3ProducerKafkaMessage.populate(type, retailerMoniker, s3fileName, bucket, provider, fileFormat, carrierMoniker);
String queueMsgStr= mapper.writeValueAsString(queueMsg);
Message out=exchange.getOut();
out.setBody(queueMsgStr, String.class);
System.out.println(queueMsgStr);
}
Is there a better way to do this. My main concern is that if there is an error while writing to kafka, then the file is still uploaded to S3. Is there a better setup where maybe i can do this in a transaction.
Use Case 2:
What I may also want to do is instead of putting a single message on kafka, for each file uploaded to S3, is batch the meta info into a single message and put on to kafka, to process the files in the current ftp poll as a batch (maybe storing all meta info in a list and then on CamelBatchComplete send message to kafka) .
Re: Route setup to upload file to s3 and put message on kafka queue
Posted by Sherwin Pinto <sh...@narvar.com>.
The way it is set up is, the route is executed every 10min . If there are no files the processors are not invoked and no action is taken.
> On Jul 17, 2016, at 11:24 AM, souciance <so...@gmail.com> wrote:
>
> I haven't used the CamelBatchComplete flag but that sounds reasonable. The
> only thing I guess is to test what happens if there are no files do
> download. Should you continue polling or stop immediately?
>
> On Sun, Jul 17, 2016 at 7:52 PM, Sherwin Pinto [via Camel] <
> ml-node+s465427n5785195h8@n5.nabble.com <ma...@n5.nabble.com>> wrote:
>
>> Hi !!
>>
>> Thank you for your input !!! I will try out your suggestions. They seem to
>> be exactly what i was looking for.
>>
>> Regarding the batch processing, I am using the FTP Idempotent Consumer
>> with a quartz scheduler. When the ftp consumer polls and downloads the
>> files, I was hoping that the CamelBatchComplete flag would be set after the
>> last file is downloaded and that would be the signal to send a message to
>> Kafka (In my processor I could maintain a list of meta info of file). I
>> haven’t looked into aggregators yet , maybe I could aggregate messages in a
>> list and then send message to kafka ?
>>
>> Thanks Again
>>
>> Regards
>>
>> Sherwin
>>
>>> On Jul 17, 2016, at 10:43 AM, souciance <[hidden email]
>> <http:///user/SendEmail.jtp?type=node&node=5785195&i=0 <http://user/SendEmail.jtp?type=node&node=5785195&i=0>>> wrote:
>>>
>>> I have not worked with S3 or Kafka so cannot help you much there but I
>>> would say the general setup of your route looks ok.
>>>
>>> The only thing is, if you are worried about the metadata being sent to
>>> Kafka if something goes wrong with the S3 part, then you need to put the
>> S3
>>> endpoint in a doTry() doCatch() statement and catch possible exceptions
>> and
>>> then send the exchange to an error handler.
>>>
>>> Alternatively, add a choice().when(<only_when_s3_ok>) and only when the
>>> predicate is true do you send the metadata to kafka.
>>>
>>> If you are going to do batch processing, how do you know when the batch
>>> starts and when the batch finishes?
>>>
>>> On Sun, Jul 17, 2016 at 7:18 PM, Sherwin Pinto [via Camel] <
>>> [hidden email] <http:///user/SendEmail.jtp?type=node&node=5785195&i=1 <http://user/SendEmail.jtp?type=node&node=5785195&i=1>>
>> <mailto:[hidden email]
>> <http:///user/SendEmail.jtp?type=node&node=5785195&i=2 <http://user/SendEmail.jtp?type=node&node=5785195&i=2>>>> wrote:
>>>
>>>> Hi All,
>>>>
>>>> Can someone help me with this ?
>>>>
>>>> Thanks
>>>>
>>>> Sherwin
>>>>
>>>>> On Jul 14, 2016, at 1:18 PM, Sherwin Pinto <[hidden email]
>>>> <http:///user/SendEmail.jtp?type=node&node=5785193&i=0 <http://user/SendEmail.jtp?type=node&node=5785193&i=0> <
>> http://user/SendEmail.jtp?type=node&node=5785193&i=0 <http://user/SendEmail.jtp?type=node&node=5785193&i=0>>>> wrote:
>>>>>
>>>>> HI All,
>>>>>
>>>>> My use case is as follows
>>>>>
>>>>> Use case 1:
>>>>> 1. Pole ftp server for file.
>>>>> 2. Upload file to S3
>>>>> 3. Put message on to kafka with file meta info i. e filename, S3
>> bucket
>>>> etc. This information is then used by a separate java process to
>> download
>>>> file and process it.
>>>>>
>>>>> I have a working prototype but want to make sure I am doing this
>>>> correctly. Here is the skeleton of the route config.
>>>>>
>>>>> from(config.getCamelFTPConsumerString()).routeId(config.getRouteId())
>>>>>
>>>>
>> .idempotentConsumer(ExpressionBuilder.headerExpression(Exchange.FILE_NAME),
>>>>> JpaMessageIdRepository.jpaMessageIdRepository(postgresEntityManager,
>>>> "FLAT_FILE_PROCESSOR"))
>>>>> .eager(true).process(myProcessor)
>>>>> .to(s3Config.getCamelS3Configuration())
>>>>> .process(kafkaProducerMessageProcessor)
>>>>> .to(kafkaConfig.getCamelKafkaConfig())
>>>>> .log("EventProc - Uploaded file ${file:name} to S3”);
>>>>>
>>>>>
>>>>> I am using a pipeline, the file is first uploaded to S3, then
>>>> kafkaProducerMessageProcessor modifies the message and replaces the
>> output
>>>> with the a JSON string that represents the kafka message which is then
>>>> routed to kafka
>>>>>
>>>>> public void process(Exchange exchange) throws Exception {
>>>>> String
>>>>
>> s3fileName=exchange.getIn().getHeader(Exchange.FILE_NAME_PRODUCED).toString();
>>
>>>>
>>>>>
>>>>> exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, null);
>>>>>
>>>>> QueueMessage<S3FileLink> queueMsg=
>> S3ProducerKafkaMessage.populate(type,
>>>> retailerMoniker, s3fileName, bucket, provider, fileFormat,
>> carrierMoniker);
>>>>>
>>>>> String queueMsgStr= mapper.writeValueAsString(queueMsg);
>>>>>
>>>>> Message out=exchange.getOut();
>>>>>
>>>>> out.setBody(queueMsgStr, String.class);
>>>>>
>>>>> System.out.println(queueMsgStr);
>>>>> }
>>>>>
>>>>> Is there a better way to do this. My main concern is that if there is
>> an
>>>> error while writing to kafka, then the file is still uploaded to S3. Is
>>>> there a better setup where maybe i can do this in a transaction.
>>>>>
>>>>> Use Case 2:
>>>>>
>>>>> What I may also want to do is instead of putting a single message on
>>>> kafka, for each file uploaded to S3, is batch the meta info into a
>> single
>>>> message and put on to kafka, to process the files in the current ftp
>> poll
>>>> as a batch (maybe storing all meta info in a list and then on
>>>> CamelBatchComplete send message to kafka) .
>>>>
>>>>
>>>>
>>>> ------------------------------
>>>> If you reply to this email, your message will be added to the
>> discussion
>>>> below:
>>>>
>>>>
>> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785193.html <http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785193.html>
>> <
>> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785193.html <http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785193.html>>
>>
>>>> To start a new topic under Camel - Users, email
>>>> [hidden email] <http:///user/SendEmail.jtp?type=node&node=5785195&i=3 <http://user/SendEmail.jtp?type=node&node=5785195&i=3>>
>> <mailto:[hidden email]
>> <http:///user/SendEmail.jtp?type=node&node=5785195&i=4 <http://user/SendEmail.jtp?type=node&node=5785195&i=4>>>
>>>> To unsubscribe from Camel - Users, click here
>>>> < href="" target="_top" rel="nofollow" link="external">
>>>> .
>>>> NAML
>>>> <
>> http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>
>> <
>> http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>>>
>>
>>>>
>>>
>>>
>>>
>>>
>>> --
>>> View this message in context:
>> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785194.html <http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785194.html>
>> <
>> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785194.html <http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785194.html>>
>>
>>> Sent from the Camel - Users mailing list archive at Nabble.com <http://nabble.com/> <
>> http://nabble.com/ <http://nabble.com/>>.
>>
>>
>>
>> ------------------------------
>> If you reply to this email, your message will be added to the discussion
>> below:
>>
>> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785195.html <http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785195.html>
>> To start a new topic under Camel - Users, email
>> ml-node+s465427n465428h31@n5.nabble.com <ma...@n5.nabble.com>
>> To unsubscribe from Camel - Users, click here
>> <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=465428&code=c291Y2lhbmNlLmVxZGFtLnJhc2h0aUBnbWFpbC5jb218NDY1NDI4fDE1MzI5MTE2NTY= <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=465428&code=c291Y2lhbmNlLmVxZGFtLnJhc2h0aUBnbWFpbC5jb218NDY1NDI4fDE1MzI5MTE2NTY=>>
>> .
>> NAML
>> <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>>
>>
>
>
>
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785196.html <http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785196.html>
> Sent from the Camel - Users mailing list archive at Nabble.com <http://nabble.com/>.
Re: Route setup to upload file to s3 and put message on kafka queue
Posted by souciance <so...@gmail.com>.
I haven't used the CamelBatchComplete flag but that sounds reasonable. The
only thing I guess is to test what happens if there are no files do
download. Should you continue polling or stop immediately?
On Sun, Jul 17, 2016 at 7:52 PM, Sherwin Pinto [via Camel] <
ml-node+s465427n5785195h8@n5.nabble.com> wrote:
> Hi !!
>
> Thank you for your input !!! I will try out your suggestions. They seem to
> be exactly what i was looking for.
>
> Regarding the batch processing, I am using the FTP Idempotent Consumer
> with a quartz scheduler. When the ftp consumer polls and downloads the
> files, I was hoping that the CamelBatchComplete flag would be set after the
> last file is downloaded and that would be the signal to send a message to
> Kafka (In my processor I could maintain a list of meta info of file). I
> haven’t looked into aggregators yet , maybe I could aggregate messages in a
> list and then send message to kafka ?
>
> Thanks Again
>
> Regards
>
> Sherwin
>
> > On Jul 17, 2016, at 10:43 AM, souciance <[hidden email]
> <http:///user/SendEmail.jtp?type=node&node=5785195&i=0>> wrote:
> >
> > I have not worked with S3 or Kafka so cannot help you much there but I
> > would say the general setup of your route looks ok.
> >
> > The only thing is, if you are worried about the metadata being sent to
> > Kafka if something goes wrong with the S3 part, then you need to put the
> S3
> > endpoint in a doTry() doCatch() statement and catch possible exceptions
> and
> > then send the exchange to an error handler.
> >
> > Alternatively, add a choice().when(<only_when_s3_ok>) and only when the
> > predicate is true do you send the metadata to kafka.
> >
> > If you are going to do batch processing, how do you know when the batch
> > starts and when the batch finishes?
> >
> > On Sun, Jul 17, 2016 at 7:18 PM, Sherwin Pinto [via Camel] <
> > [hidden email] <http:///user/SendEmail.jtp?type=node&node=5785195&i=1>
> <mailto:[hidden email]
> <http:///user/SendEmail.jtp?type=node&node=5785195&i=2>>> wrote:
> >
> >> Hi All,
> >>
> >> Can someone help me with this ?
> >>
> >> Thanks
> >>
> >> Sherwin
> >>
> >>> On Jul 14, 2016, at 1:18 PM, Sherwin Pinto <[hidden email]
> >> <http:///user/SendEmail.jtp?type=node&node=5785193&i=0 <
> http://user/SendEmail.jtp?type=node&node=5785193&i=0>>> wrote:
> >>>
> >>> HI All,
> >>>
> >>> My use case is as follows
> >>>
> >>> Use case 1:
> >>> 1. Pole ftp server for file.
> >>> 2. Upload file to S3
> >>> 3. Put message on to kafka with file meta info i. e filename, S3
> bucket
> >> etc. This information is then used by a separate java process to
> download
> >> file and process it.
> >>>
> >>> I have a working prototype but want to make sure I am doing this
> >> correctly. Here is the skeleton of the route config.
> >>>
> >>> from(config.getCamelFTPConsumerString()).routeId(config.getRouteId())
> >>>
> >>
> .idempotentConsumer(ExpressionBuilder.headerExpression(Exchange.FILE_NAME),
> >>> JpaMessageIdRepository.jpaMessageIdRepository(postgresEntityManager,
> >> "FLAT_FILE_PROCESSOR"))
> >>> .eager(true).process(myProcessor)
> >>> .to(s3Config.getCamelS3Configuration())
> >>> .process(kafkaProducerMessageProcessor)
> >>> .to(kafkaConfig.getCamelKafkaConfig())
> >>> .log("EventProc - Uploaded file ${file:name} to S3”);
> >>>
> >>>
> >>> I am using a pipeline, the file is first uploaded to S3, then
> >> kafkaProducerMessageProcessor modifies the message and replaces the
> output
> >> with the a JSON string that represents the kafka message which is then
> >> routed to kafka
> >>>
> >>> public void process(Exchange exchange) throws Exception {
> >>> String
> >>
> s3fileName=exchange.getIn().getHeader(Exchange.FILE_NAME_PRODUCED).toString();
>
> >>
> >>>
> >>> exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, null);
> >>>
> >>> QueueMessage<S3FileLink> queueMsg=
> S3ProducerKafkaMessage.populate(type,
> >> retailerMoniker, s3fileName, bucket, provider, fileFormat,
> carrierMoniker);
> >>>
> >>> String queueMsgStr= mapper.writeValueAsString(queueMsg);
> >>>
> >>> Message out=exchange.getOut();
> >>>
> >>> out.setBody(queueMsgStr, String.class);
> >>>
> >>> System.out.println(queueMsgStr);
> >>> }
> >>>
> >>> Is there a better way to do this. My main concern is that if there is
> an
> >> error while writing to kafka, then the file is still uploaded to S3. Is
> >> there a better setup where maybe i can do this in a transaction.
> >>>
> >>> Use Case 2:
> >>>
> >>> What I may also want to do is instead of putting a single message on
> >> kafka, for each file uploaded to S3, is batch the meta info into a
> single
> >> message and put on to kafka, to process the files in the current ftp
> poll
> >> as a batch (maybe storing all meta info in a list and then on
> >> CamelBatchComplete send message to kafka) .
> >>
> >>
> >>
> >> ------------------------------
> >> If you reply to this email, your message will be added to the
> discussion
> >> below:
> >>
> >>
> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785193.html
> <
> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785193.html>
>
> >> To start a new topic under Camel - Users, email
> >> [hidden email] <http:///user/SendEmail.jtp?type=node&node=5785195&i=3>
> <mailto:[hidden email]
> <http:///user/SendEmail.jtp?type=node&node=5785195&i=4>>
> >> To unsubscribe from Camel - Users, click here
> >> < href="" target="_top" rel="nofollow" link="external">
> >> .
> >> NAML
> >> <
> http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml
> <
> http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>>
>
> >>
> >
> >
> >
> >
> > --
> > View this message in context:
> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785194.html
> <
> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785194.html>
>
> > Sent from the Camel - Users mailing list archive at Nabble.com <
> http://nabble.com/>.
>
>
>
> ------------------------------
> If you reply to this email, your message will be added to the discussion
> below:
>
> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785195.html
> To start a new topic under Camel - Users, email
> ml-node+s465427n465428h31@n5.nabble.com
> To unsubscribe from Camel - Users, click here
> <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=465428&code=c291Y2lhbmNlLmVxZGFtLnJhc2h0aUBnbWFpbC5jb218NDY1NDI4fDE1MzI5MTE2NTY=>
> .
> NAML
> <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>
>
--
View this message in context: http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785196.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Route setup to upload file to s3 and put message on kafka queue
Posted by Sherwin Pinto <sh...@narvar.com>.
Hi !!
Thank you for your input !!! I will try out your suggestions. They seem to be exactly what i was looking for.
Regarding the batch processing, I am using the FTP Idempotent Consumer with a quartz scheduler. When the ftp consumer polls and downloads the files, I was hoping that the CamelBatchComplete flag would be set after the last file is downloaded and that would be the signal to send a message to Kafka (In my processor I could maintain a list of meta info of file). I haven’t looked into aggregators yet , maybe I could aggregate messages in a list and then send message to kafka ?
Thanks Again
Regards
Sherwin
> On Jul 17, 2016, at 10:43 AM, souciance <so...@gmail.com> wrote:
>
> I have not worked with S3 or Kafka so cannot help you much there but I
> would say the general setup of your route looks ok.
>
> The only thing is, if you are worried about the metadata being sent to
> Kafka if something goes wrong with the S3 part, then you need to put the S3
> endpoint in a doTry() doCatch() statement and catch possible exceptions and
> then send the exchange to an error handler.
>
> Alternatively, add a choice().when(<only_when_s3_ok>) and only when the
> predicate is true do you send the metadata to kafka.
>
> If you are going to do batch processing, how do you know when the batch
> starts and when the batch finishes?
>
> On Sun, Jul 17, 2016 at 7:18 PM, Sherwin Pinto [via Camel] <
> ml-node+s465427n5785193h46@n5.nabble.com <ma...@n5.nabble.com>> wrote:
>
>> Hi All,
>>
>> Can someone help me with this ?
>>
>> Thanks
>>
>> Sherwin
>>
>>> On Jul 14, 2016, at 1:18 PM, Sherwin Pinto <[hidden email]
>> <http:///user/SendEmail.jtp?type=node&node=5785193&i=0 <http://user/SendEmail.jtp?type=node&node=5785193&i=0>>> wrote:
>>>
>>> HI All,
>>>
>>> My use case is as follows
>>>
>>> Use case 1:
>>> 1. Pole ftp server for file.
>>> 2. Upload file to S3
>>> 3. Put message on to kafka with file meta info i. e filename, S3 bucket
>> etc. This information is then used by a separate java process to download
>> file and process it.
>>>
>>> I have a working prototype but want to make sure I am doing this
>> correctly. Here is the skeleton of the route config.
>>>
>>> from(config.getCamelFTPConsumerString()).routeId(config.getRouteId())
>>>
>> .idempotentConsumer(ExpressionBuilder.headerExpression(Exchange.FILE_NAME),
>>> JpaMessageIdRepository.jpaMessageIdRepository(postgresEntityManager,
>> "FLAT_FILE_PROCESSOR"))
>>> .eager(true).process(myProcessor)
>>> .to(s3Config.getCamelS3Configuration())
>>> .process(kafkaProducerMessageProcessor)
>>> .to(kafkaConfig.getCamelKafkaConfig())
>>> .log("EventProc - Uploaded file ${file:name} to S3”);
>>>
>>>
>>> I am using a pipeline, the file is first uploaded to S3, then
>> kafkaProducerMessageProcessor modifies the message and replaces the output
>> with the a JSON string that represents the kafka message which is then
>> routed to kafka
>>>
>>> public void process(Exchange exchange) throws Exception {
>>> String
>> s3fileName=exchange.getIn().getHeader(Exchange.FILE_NAME_PRODUCED).toString();
>>
>>>
>>> exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, null);
>>>
>>> QueueMessage<S3FileLink> queueMsg= S3ProducerKafkaMessage.populate(type,
>> retailerMoniker, s3fileName, bucket, provider, fileFormat, carrierMoniker);
>>>
>>> String queueMsgStr= mapper.writeValueAsString(queueMsg);
>>>
>>> Message out=exchange.getOut();
>>>
>>> out.setBody(queueMsgStr, String.class);
>>>
>>> System.out.println(queueMsgStr);
>>> }
>>>
>>> Is there a better way to do this. My main concern is that if there is an
>> error while writing to kafka, then the file is still uploaded to S3. Is
>> there a better setup where maybe i can do this in a transaction.
>>>
>>> Use Case 2:
>>>
>>> What I may also want to do is instead of putting a single message on
>> kafka, for each file uploaded to S3, is batch the meta info into a single
>> message and put on to kafka, to process the files in the current ftp poll
>> as a batch (maybe storing all meta info in a list and then on
>> CamelBatchComplete send message to kafka) .
>>
>>
>>
>> ------------------------------
>> If you reply to this email, your message will be added to the discussion
>> below:
>>
>> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785193.html <http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785193.html>
>> To start a new topic under Camel - Users, email
>> ml-node+s465427n465428h31@n5.nabble.com <ma...@n5.nabble.com>
>> To unsubscribe from Camel - Users, click here
>> <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=465428&code=c291Y2lhbmNlLmVxZGFtLnJhc2h0aUBnbWFpbC5jb218NDY1NDI4fDE1MzI5MTE2NTY= <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=465428&code=c291Y2lhbmNlLmVxZGFtLnJhc2h0aUBnbWFpbC5jb218NDY1NDI4fDE1MzI5MTE2NTY=>>
>> .
>> NAML
>> <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>>
>>
>
>
>
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785194.html <http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785194.html>
> Sent from the Camel - Users mailing list archive at Nabble.com <http://nabble.com/>.
Re: Route setup to upload file to s3 and put message on kafka queue
Posted by souciance <so...@gmail.com>.
I have not worked with S3 or Kafka so cannot help you much there but I
would say the general setup of your route looks ok.
The only thing is, if you are worried about the metadata being sent to
Kafka if something goes wrong with the S3 part, then you need to put the S3
endpoint in a doTry() doCatch() statement and catch possible exceptions and
then send the exchange to an error handler.
Alternatively, add a choice().when(<only_when_s3_ok>) and only when the
predicate is true do you send the metadata to kafka.
If you are going to do batch processing, how do you know when the batch
starts and when the batch finishes?
On Sun, Jul 17, 2016 at 7:18 PM, Sherwin Pinto [via Camel] <
ml-node+s465427n5785193h46@n5.nabble.com> wrote:
> Hi All,
>
> Can someone help me with this ?
>
> Thanks
>
> Sherwin
>
> > On Jul 14, 2016, at 1:18 PM, Sherwin Pinto <[hidden email]
> <http:///user/SendEmail.jtp?type=node&node=5785193&i=0>> wrote:
> >
> > HI All,
> >
> > My use case is as follows
> >
> > Use case 1:
> > 1. Pole ftp server for file.
> > 2. Upload file to S3
> > 3. Put message on to kafka with file meta info i. e filename, S3 bucket
> etc. This information is then used by a separate java process to download
> file and process it.
> >
> > I have a working prototype but want to make sure I am doing this
> correctly. Here is the skeleton of the route config.
> >
> > from(config.getCamelFTPConsumerString()).routeId(config.getRouteId())
> >
> .idempotentConsumer(ExpressionBuilder.headerExpression(Exchange.FILE_NAME),
> > JpaMessageIdRepository.jpaMessageIdRepository(postgresEntityManager,
> "FLAT_FILE_PROCESSOR"))
> > .eager(true).process(myProcessor)
> > .to(s3Config.getCamelS3Configuration())
> > .process(kafkaProducerMessageProcessor)
> > .to(kafkaConfig.getCamelKafkaConfig())
> > .log("EventProc - Uploaded file ${file:name} to S3”);
> >
> >
> > I am using a pipeline, the file is first uploaded to S3, then
> kafkaProducerMessageProcessor modifies the message and replaces the output
> with the a JSON string that represents the kafka message which is then
> routed to kafka
> >
> > public void process(Exchange exchange) throws Exception {
> > String
> s3fileName=exchange.getIn().getHeader(Exchange.FILE_NAME_PRODUCED).toString();
>
> >
> > exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, null);
> >
> > QueueMessage<S3FileLink> queueMsg= S3ProducerKafkaMessage.populate(type,
> retailerMoniker, s3fileName, bucket, provider, fileFormat, carrierMoniker);
> >
> > String queueMsgStr= mapper.writeValueAsString(queueMsg);
> >
> > Message out=exchange.getOut();
> >
> > out.setBody(queueMsgStr, String.class);
> >
> > System.out.println(queueMsgStr);
> > }
> >
> > Is there a better way to do this. My main concern is that if there is an
> error while writing to kafka, then the file is still uploaded to S3. Is
> there a better setup where maybe i can do this in a transaction.
> >
> > Use Case 2:
> >
> > What I may also want to do is instead of putting a single message on
> kafka, for each file uploaded to S3, is batch the meta info into a single
> message and put on to kafka, to process the files in the current ftp poll
> as a batch (maybe storing all meta info in a list and then on
> CamelBatchComplete send message to kafka) .
>
>
>
> ------------------------------
> If you reply to this email, your message will be added to the discussion
> below:
>
> http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785193.html
> To start a new topic under Camel - Users, email
> ml-node+s465427n465428h31@n5.nabble.com
> To unsubscribe from Camel - Users, click here
> <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=465428&code=c291Y2lhbmNlLmVxZGFtLnJhc2h0aUBnbWFpbC5jb218NDY1NDI4fDE1MzI5MTE2NTY=>
> .
> NAML
> <http://camel.465427.n5.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>
>
--
View this message in context: http://camel.465427.n5.nabble.com/Route-setup-to-upload-file-to-s3-and-put-message-on-kafka-queue-tp5785115p5785194.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Route setup to upload file to s3 and put message on kafka queue
Posted by Sherwin Pinto <sh...@narvar.com>.
Hi All,
Can someone help me with this ?
Thanks
Sherwin
> On Jul 14, 2016, at 1:18 PM, Sherwin Pinto <sh...@narvar.com> wrote:
>
> HI All,
>
> My use case is as follows
>
> Use case 1:
> 1. Pole ftp server for file.
> 2. Upload file to S3
> 3. Put message on to kafka with file meta info i. e filename, S3 bucket etc. This information is then used by a separate java process to download file and process it.
>
> I have a working prototype but want to make sure I am doing this correctly. Here is the skeleton of the route config.
>
> from(config.getCamelFTPConsumerString()).routeId(config.getRouteId())
> .idempotentConsumer(ExpressionBuilder.headerExpression(Exchange.FILE_NAME),
> JpaMessageIdRepository.jpaMessageIdRepository(postgresEntityManager, "FLAT_FILE_PROCESSOR"))
> .eager(true).process(myProcessor)
> .to(s3Config.getCamelS3Configuration())
> .process(kafkaProducerMessageProcessor)
> .to(kafkaConfig.getCamelKafkaConfig())
> .log("EventProc - Uploaded file ${file:name} to S3”);
>
>
> I am using a pipeline, the file is first uploaded to S3, then kafkaProducerMessageProcessor modifies the message and replaces the output with the a JSON string that represents the kafka message which is then routed to kafka
>
> public void process(Exchange exchange) throws Exception {
> String s3fileName=exchange.getIn().getHeader(Exchange.FILE_NAME_PRODUCED).toString();
>
> exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, null);
>
> QueueMessage<S3FileLink> queueMsg= S3ProducerKafkaMessage.populate(type, retailerMoniker, s3fileName, bucket, provider, fileFormat, carrierMoniker);
>
> String queueMsgStr= mapper.writeValueAsString(queueMsg);
>
> Message out=exchange.getOut();
>
> out.setBody(queueMsgStr, String.class);
>
> System.out.println(queueMsgStr);
> }
>
> Is there a better way to do this. My main concern is that if there is an error while writing to kafka, then the file is still uploaded to S3. Is there a better setup where maybe i can do this in a transaction.
>
> Use Case 2:
>
> What I may also want to do is instead of putting a single message on kafka, for each file uploaded to S3, is batch the meta info into a single message and put on to kafka, to process the files in the current ftp poll as a batch (maybe storing all meta info in a list and then on CamelBatchComplete send message to kafka) .