david-streamlio commented on a change in pull request #9943:
URL: https://github.com/apache/pulsar/pull/9943#discussion_r596341448
##########
File path:
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
##########
@@ -736,8 +737,14 @@ private void setupOutput(ContextImpl contextImpl) throws
Exception {
pulsarSinkConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.valueOf(
this.instanceConfig.getFunctionDetails().getProcessingGuarantees().name()));
pulsarSinkConfig.setTopic(sinkSpec.getTopic());
- pulsarSinkConfig.setForwardSourceMessageProperty(
+
+ if
(this.instanceConfig.getFunctionDetails().getComponentType()==
ComponentType.SOURCE) {
+ // Forward record level properties for all sources by default
+ pulsarSinkConfig.setForwardSourceMessageProperty(true);
+ } else if
(this.instanceConfig.getFunctionDetails().getComponentType() ==
ComponentType.SINK) {
Review comment:
A more comprehensive solution would be to add a
`ForwardSourceMessageProperty` to the configuration for ALL component types and
use a default value of `true` for all type? Seems like from a consistency
perspective having all the components behave exactly the same w.r.t to forward
properties but leave an option for the user to override this behavior as need.
Thoughts?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
