Re: Sending exchange through Pulsar endpoint, exchange properties disappear
Hi Steve -
I think you’re talking about consuming the property from the Pulsar topic
rather than “within” the Camel route.
If you want the properties to make it into the Pulsar queue, rather than
just in Camel, you’ll need to set PROPERTIES_OUT (add to the key-value
pair) in the header rather than in the exchange property. See
https://camel.apache.org/components/3.20.x/pulsar-component.html#_message_headers
and
https://pulsar.apache.org/docs/2.11.x/concepts-messaging/#messages.
(Should you want to read out this properties from Pulsar to Camel, you want
to reference PROPERTIES.)
On Fri, Apr 14, 2023 at 2:07 PM Steve973 wrote:
> Ah, I just looked at the code, and it doesn't grab any user headers. The
> closest thing to user properties or headers that I can find is a field in
> the PulsarMessageHeaders class:
> @Metadata(label = "producer", description = "The properties of the message
> to add.", javaType = "Map")
> String PROPERTIES_OUT = "CamelPulsarProducerMessageProperties";
>
> Am I missing something?
>
>
> On Fri, Apr 14, 2023 at 4:52 PM Steve973 wrote:
>
> > Claus,
> >
> > I switched to using .withHeader() calls instead of .withProperty(), but I
> > am still not seeing my headers when I read the message from Pulsar. How
> > does the Pulsar component attempt to keep track of the headers?
> >
> >
> > On Fri, Apr 14, 2023 at 10:47 AM Steve973 wrote:
> >
> >> Claus,
> >>
> >> Thanks for that. I suppose I was a bit confused about the difference.
> >> In 2012, you said (on SO):
> >>
> >>> One distinction not mentioned by Ben and Petter is that properties are
> >>> safely stored for the entire duration of the processing of the message
> in
> >>> Camel. In contrast, headers are part of the message protocol, and may
> not
> >>> be propagated during routing. For example, JMS has limitations on what
> >>> you can store as headers etc.
> >>
> >>
> >> On the same thread, I see Petter Nordlander claim:
> >>
> >>> Properties and headers are pretty much the same. Headers are, however,
> >>> converted to/from protocol specific headers on certain components,
> such as
> >>> Jms. So,
> >>>
> >>>- Meta data inside a route: properties
> >>>- Meta data to/from outside: headers
> >>>
> >>>
> >> Is this distinction because what happens with pulsar is technically
> >> "outside" of the Exchange itself? So the specific protocols handle just
> >> what is supplied in the body and headers? And the headers have to get
> >> mapped to something in the specific protocols that are used, and then
> they
> >> will get mapped back into the message when read again by Camel when
> >> consuming from that protocol?
> >>
> >> Hopefully I'm not even more confused now!
> >>
> >> On Fri, Apr 14, 2023 at 5:03 AM Claus Ibsen
> >> wrote:
> >>
> >>> Hi
> >>>
> >>> This is expected, exchange properties are local only to the current
> >>> Exchange.
> >>> If you send data over network protocols like pulsar, kafka, jms then
> >>> the payload is only from the message (eg body and headers)
> >>>
> >>>
> >>>
> >>> On Thu, Apr 13, 2023 at 11:24 PM Steve973 wrote:
> >>>
> >>> > Hello. I am sending an exchange with a FluentProducerTemplate to a
> >>> pulsar
> >>> > endpoint. When a route reads from that endpoint, the exchange
> property
> >>> > that I set is no longer there. This is how I am sending the message:
> >>> >
> >>> > Exchange exchange = ExchangeBuilder.anExchange(camelContext)
> >>> > .withProperty("command", MessageTypes.GENERATE_NUMBERS_COMMAND)
> >>> > .withBody(generateNumbersCommandMessage.toByteArray())
> >>> > .build();
> >>> > fluentProducerTemplate.withExchange(exchange)
> >>> > .to(commandUri)
> >>> > .send();
> >>> >
> >>> > I am reading from the same URI to feed a route where I log the
> >>> exchange:
> >>> >
> >>> > from(mainRouterConfig.commandEntrypoint())
> >>> >
> .to("log:to.dynamic-router.command.route?showAll=true=true")
> >>> > .to("%s:%s?recipientMode=%s".formatted(
> >>> > COMPONENT_SCHEME, mainRouterConfig.routingChannel(), mainRouterConfig
> >>> > .recipientMode()));
> >>> >
> >>> > The exchange property is absent in the log message:
> >>> >
> >>> > Id: 453CF424AB353C4-
> >>> > ExchangePattern: InOnly
> >>> > Properties:
> >>> >
> >>> >
> >>>
> {CamelToEndpoint=log://to.dynamic-router.command.route?multiline=true=true}
> >>> >
> >>> > Headers: {CamelMessageTimestamp=1681420673229,
> >>> > CamelPulsarRedeliveryCount=0, event_time=0, key=null, key_bytes=null,
> >>> > message_id=0:0:-1, producer_name=standalone-0-11, properties={},
> >>> > publish_time=1681420673229
> >>> > , sequence_id=0,
> >>> > topic_name=non-persistent://public/default/numbers/command}
> >>> > BodyType: byte[]
> >>> > Body: generateNumberstypesequentialstart0limit100
> >>> >
> >>> > Am I doing anything obviously wrong? Or is the Pulsar component
> eating
> >>> > exchange properties?
> >>> >
> >>> > Thanks,
> >>> > Steve
> >>> >
> >>>
> >>>
> >>> --
> >>> Claus Ibsen
> >>> -
> >>>
Re: Sending exchange through Pulsar endpoint, exchange properties disappear
Ah, I just looked at the code, and it doesn't grab any user headers. The
closest thing to user properties or headers that I can find is a field in
the PulsarMessageHeaders class:
@Metadata(label = "producer", description = "The properties of the message
to add.", javaType = "Map")
String PROPERTIES_OUT = "CamelPulsarProducerMessageProperties";
Am I missing something?
On Fri, Apr 14, 2023 at 4:52 PM Steve973 wrote:
> Claus,
>
> I switched to using .withHeader() calls instead of .withProperty(), but I
> am still not seeing my headers when I read the message from Pulsar. How
> does the Pulsar component attempt to keep track of the headers?
>
>
> On Fri, Apr 14, 2023 at 10:47 AM Steve973 wrote:
>
>> Claus,
>>
>> Thanks for that. I suppose I was a bit confused about the difference.
>> In 2012, you said (on SO):
>>
>>> One distinction not mentioned by Ben and Petter is that properties are
>>> safely stored for the entire duration of the processing of the message in
>>> Camel. In contrast, headers are part of the message protocol, and may not
>>> be propagated during routing. For example, JMS has limitations on what
>>> you can store as headers etc.
>>
>>
>> On the same thread, I see Petter Nordlander claim:
>>
>>> Properties and headers are pretty much the same. Headers are, however,
>>> converted to/from protocol specific headers on certain components, such as
>>> Jms. So,
>>>
>>>- Meta data inside a route: properties
>>>- Meta data to/from outside: headers
>>>
>>>
>> Is this distinction because what happens with pulsar is technically
>> "outside" of the Exchange itself? So the specific protocols handle just
>> what is supplied in the body and headers? And the headers have to get
>> mapped to something in the specific protocols that are used, and then they
>> will get mapped back into the message when read again by Camel when
>> consuming from that protocol?
>>
>> Hopefully I'm not even more confused now!
>>
>> On Fri, Apr 14, 2023 at 5:03 AM Claus Ibsen
>> wrote:
>>
>>> Hi
>>>
>>> This is expected, exchange properties are local only to the current
>>> Exchange.
>>> If you send data over network protocols like pulsar, kafka, jms then
>>> the payload is only from the message (eg body and headers)
>>>
>>>
>>>
>>> On Thu, Apr 13, 2023 at 11:24 PM Steve973 wrote:
>>>
>>> > Hello. I am sending an exchange with a FluentProducerTemplate to a
>>> pulsar
>>> > endpoint. When a route reads from that endpoint, the exchange property
>>> > that I set is no longer there. This is how I am sending the message:
>>> >
>>> > Exchange exchange = ExchangeBuilder.anExchange(camelContext)
>>> > .withProperty("command", MessageTypes.GENERATE_NUMBERS_COMMAND)
>>> > .withBody(generateNumbersCommandMessage.toByteArray())
>>> > .build();
>>> > fluentProducerTemplate.withExchange(exchange)
>>> > .to(commandUri)
>>> > .send();
>>> >
>>> > I am reading from the same URI to feed a route where I log the
>>> exchange:
>>> >
>>> > from(mainRouterConfig.commandEntrypoint())
>>> > .to("log:to.dynamic-router.command.route?showAll=true=true")
>>> > .to("%s:%s?recipientMode=%s".formatted(
>>> > COMPONENT_SCHEME, mainRouterConfig.routingChannel(), mainRouterConfig
>>> > .recipientMode()));
>>> >
>>> > The exchange property is absent in the log message:
>>> >
>>> > Id: 453CF424AB353C4-
>>> > ExchangePattern: InOnly
>>> > Properties:
>>> >
>>> >
>>> {CamelToEndpoint=log://to.dynamic-router.command.route?multiline=true=true}
>>> >
>>> > Headers: {CamelMessageTimestamp=1681420673229,
>>> > CamelPulsarRedeliveryCount=0, event_time=0, key=null, key_bytes=null,
>>> > message_id=0:0:-1, producer_name=standalone-0-11, properties={},
>>> > publish_time=1681420673229
>>> > , sequence_id=0,
>>> > topic_name=non-persistent://public/default/numbers/command}
>>> > BodyType: byte[]
>>> > Body: generateNumberstypesequentialstart0limit100
>>> >
>>> > Am I doing anything obviously wrong? Or is the Pulsar component eating
>>> > exchange properties?
>>> >
>>> > Thanks,
>>> > Steve
>>> >
>>>
>>>
>>> --
>>> Claus Ibsen
>>> -
>>> @davsclaus
>>> Camel in Action 2: https://www.manning.com/ibsen2
>>>
>>
Re: Sending exchange through Pulsar endpoint, exchange properties disappear
Claus,
I switched to using .withHeader() calls instead of .withProperty(), but I
am still not seeing my headers when I read the message from Pulsar. How
does the Pulsar component attempt to keep track of the headers?
On Fri, Apr 14, 2023 at 10:47 AM Steve973 wrote:
> Claus,
>
> Thanks for that. I suppose I was a bit confused about the difference. In
> 2012, you said (on SO):
>
>> One distinction not mentioned by Ben and Petter is that properties are
>> safely stored for the entire duration of the processing of the message in
>> Camel. In contrast, headers are part of the message protocol, and may not
>> be propagated during routing. For example, JMS has limitations on what
>> you can store as headers etc.
>
>
> On the same thread, I see Petter Nordlander claim:
>
>> Properties and headers are pretty much the same. Headers are, however,
>> converted to/from protocol specific headers on certain components, such as
>> Jms. So,
>>
>>- Meta data inside a route: properties
>>- Meta data to/from outside: headers
>>
>>
> Is this distinction because what happens with pulsar is technically
> "outside" of the Exchange itself? So the specific protocols handle just
> what is supplied in the body and headers? And the headers have to get
> mapped to something in the specific protocols that are used, and then they
> will get mapped back into the message when read again by Camel when
> consuming from that protocol?
>
> Hopefully I'm not even more confused now!
>
> On Fri, Apr 14, 2023 at 5:03 AM Claus Ibsen wrote:
>
>> Hi
>>
>> This is expected, exchange properties are local only to the current
>> Exchange.
>> If you send data over network protocols like pulsar, kafka, jms then
>> the payload is only from the message (eg body and headers)
>>
>>
>>
>> On Thu, Apr 13, 2023 at 11:24 PM Steve973 wrote:
>>
>> > Hello. I am sending an exchange with a FluentProducerTemplate to a
>> pulsar
>> > endpoint. When a route reads from that endpoint, the exchange property
>> > that I set is no longer there. This is how I am sending the message:
>> >
>> > Exchange exchange = ExchangeBuilder.anExchange(camelContext)
>> > .withProperty("command", MessageTypes.GENERATE_NUMBERS_COMMAND)
>> > .withBody(generateNumbersCommandMessage.toByteArray())
>> > .build();
>> > fluentProducerTemplate.withExchange(exchange)
>> > .to(commandUri)
>> > .send();
>> >
>> > I am reading from the same URI to feed a route where I log the exchange:
>> >
>> > from(mainRouterConfig.commandEntrypoint())
>> > .to("log:to.dynamic-router.command.route?showAll=true=true")
>> > .to("%s:%s?recipientMode=%s".formatted(
>> > COMPONENT_SCHEME, mainRouterConfig.routingChannel(), mainRouterConfig
>> > .recipientMode()));
>> >
>> > The exchange property is absent in the log message:
>> >
>> > Id: 453CF424AB353C4-
>> > ExchangePattern: InOnly
>> > Properties:
>> >
>> >
>> {CamelToEndpoint=log://to.dynamic-router.command.route?multiline=true=true}
>> >
>> > Headers: {CamelMessageTimestamp=1681420673229,
>> > CamelPulsarRedeliveryCount=0, event_time=0, key=null, key_bytes=null,
>> > message_id=0:0:-1, producer_name=standalone-0-11, properties={},
>> > publish_time=1681420673229
>> > , sequence_id=0,
>> > topic_name=non-persistent://public/default/numbers/command}
>> > BodyType: byte[]
>> > Body: generateNumberstypesequentialstart0limit100
>> >
>> > Am I doing anything obviously wrong? Or is the Pulsar component eating
>> > exchange properties?
>> >
>> > Thanks,
>> > Steve
>> >
>>
>>
>> --
>> Claus Ibsen
>> -
>> @davsclaus
>> Camel in Action 2: https://www.manning.com/ibsen2
>>
>
Re: Sending exchange through Pulsar endpoint, exchange properties disappear
Claus,
Thanks for that. I suppose I was a bit confused about the difference. In
2012, you said (on SO):
> One distinction not mentioned by Ben and Petter is that properties are
> safely stored for the entire duration of the processing of the message in
> Camel. In contrast, headers are part of the message protocol, and may not
> be propagated during routing. For example, JMS has limitations on what
> you can store as headers etc.
On the same thread, I see Petter Nordlander claim:
> Properties and headers are pretty much the same. Headers are, however,
> converted to/from protocol specific headers on certain components, such as
> Jms. So,
>
>- Meta data inside a route: properties
>- Meta data to/from outside: headers
>
>
Is this distinction because what happens with pulsar is technically
"outside" of the Exchange itself? So the specific protocols handle just
what is supplied in the body and headers? And the headers have to get
mapped to something in the specific protocols that are used, and then they
will get mapped back into the message when read again by Camel when
consuming from that protocol?
Hopefully I'm not even more confused now!
On Fri, Apr 14, 2023 at 5:03 AM Claus Ibsen wrote:
> Hi
>
> This is expected, exchange properties are local only to the current
> Exchange.
> If you send data over network protocols like pulsar, kafka, jms then
> the payload is only from the message (eg body and headers)
>
>
>
> On Thu, Apr 13, 2023 at 11:24 PM Steve973 wrote:
>
> > Hello. I am sending an exchange with a FluentProducerTemplate to a
> pulsar
> > endpoint. When a route reads from that endpoint, the exchange property
> > that I set is no longer there. This is how I am sending the message:
> >
> > Exchange exchange = ExchangeBuilder.anExchange(camelContext)
> > .withProperty("command", MessageTypes.GENERATE_NUMBERS_COMMAND)
> > .withBody(generateNumbersCommandMessage.toByteArray())
> > .build();
> > fluentProducerTemplate.withExchange(exchange)
> > .to(commandUri)
> > .send();
> >
> > I am reading from the same URI to feed a route where I log the exchange:
> >
> > from(mainRouterConfig.commandEntrypoint())
> > .to("log:to.dynamic-router.command.route?showAll=true=true")
> > .to("%s:%s?recipientMode=%s".formatted(
> > COMPONENT_SCHEME, mainRouterConfig.routingChannel(), mainRouterConfig
> > .recipientMode()));
> >
> > The exchange property is absent in the log message:
> >
> > Id: 453CF424AB353C4-
> > ExchangePattern: InOnly
> > Properties:
> >
> >
> {CamelToEndpoint=log://to.dynamic-router.command.route?multiline=true=true}
> >
> > Headers: {CamelMessageTimestamp=1681420673229,
> > CamelPulsarRedeliveryCount=0, event_time=0, key=null, key_bytes=null,
> > message_id=0:0:-1, producer_name=standalone-0-11, properties={},
> > publish_time=1681420673229
> > , sequence_id=0,
> > topic_name=non-persistent://public/default/numbers/command}
> > BodyType: byte[]
> > Body: generateNumberstypesequentialstart0limit100
> >
> > Am I doing anything obviously wrong? Or is the Pulsar component eating
> > exchange properties?
> >
> > Thanks,
> > Steve
> >
>
>
> --
> Claus Ibsen
> -
> @davsclaus
> Camel in Action 2: https://www.manning.com/ibsen2
>
Re: Sending exchange through Pulsar endpoint, exchange properties disappear
Hi
This is expected, exchange properties are local only to the current
Exchange.
If you send data over network protocols like pulsar, kafka, jms then
the payload is only from the message (eg body and headers)
On Thu, Apr 13, 2023 at 11:24 PM Steve973 wrote:
> Hello. I am sending an exchange with a FluentProducerTemplate to a pulsar
> endpoint. When a route reads from that endpoint, the exchange property
> that I set is no longer there. This is how I am sending the message:
>
> Exchange exchange = ExchangeBuilder.anExchange(camelContext)
> .withProperty("command", MessageTypes.GENERATE_NUMBERS_COMMAND)
> .withBody(generateNumbersCommandMessage.toByteArray())
> .build();
> fluentProducerTemplate.withExchange(exchange)
> .to(commandUri)
> .send();
>
> I am reading from the same URI to feed a route where I log the exchange:
>
> from(mainRouterConfig.commandEntrypoint())
> .to("log:to.dynamic-router.command.route?showAll=true=true")
> .to("%s:%s?recipientMode=%s".formatted(
> COMPONENT_SCHEME, mainRouterConfig.routingChannel(), mainRouterConfig
> .recipientMode()));
>
> The exchange property is absent in the log message:
>
> Id: 453CF424AB353C4-
> ExchangePattern: InOnly
> Properties:
>
> {CamelToEndpoint=log://to.dynamic-router.command.route?multiline=true=true}
>
> Headers: {CamelMessageTimestamp=1681420673229,
> CamelPulsarRedeliveryCount=0, event_time=0, key=null, key_bytes=null,
> message_id=0:0:-1, producer_name=standalone-0-11, properties={},
> publish_time=1681420673229
> , sequence_id=0,
> topic_name=non-persistent://public/default/numbers/command}
> BodyType: byte[]
> Body: generateNumberstypesequentialstart0limit100
>
> Am I doing anything obviously wrong? Or is the Pulsar component eating
> exchange properties?
>
> Thanks,
> Steve
>
--
Claus Ibsen
-
@davsclaus
Camel in Action 2: https://www.manning.com/ibsen2
Sending exchange through Pulsar endpoint, exchange properties disappear
Hello. I am sending an exchange with a FluentProducerTemplate to a pulsar
endpoint. When a route reads from that endpoint, the exchange property
that I set is no longer there. This is how I am sending the message:
Exchange exchange = ExchangeBuilder.anExchange(camelContext)
.withProperty("command", MessageTypes.GENERATE_NUMBERS_COMMAND)
.withBody(generateNumbersCommandMessage.toByteArray())
.build();
fluentProducerTemplate.withExchange(exchange)
.to(commandUri)
.send();
I am reading from the same URI to feed a route where I log the exchange:
from(mainRouterConfig.commandEntrypoint())
.to("log:to.dynamic-router.command.route?showAll=true=true")
.to("%s:%s?recipientMode=%s".formatted(
COMPONENT_SCHEME, mainRouterConfig.routingChannel(), mainRouterConfig
.recipientMode()));
The exchange property is absent in the log message:
Id: 453CF424AB353C4-
ExchangePattern: InOnly
Properties:
{CamelToEndpoint=log://to.dynamic-router.command.route?multiline=true=true}
Headers: {CamelMessageTimestamp=1681420673229,
CamelPulsarRedeliveryCount=0, event_time=0, key=null, key_bytes=null,
message_id=0:0:-1, producer_name=standalone-0-11, properties={},
publish_time=1681420673229
, sequence_id=0,
topic_name=non-persistent://public/default/numbers/command}
BodyType: byte[]
Body: generateNumberstypesequentialstart0limit100
Am I doing anything obviously wrong? Or is the Pulsar component eating
exchange properties?
Thanks,
Steve
