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: I 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: us...@infra.apache.org