Nifi自定义处理器如何在内容或属性中写入结果

import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.annotation.lifecycle.OnShutdown;

import java.io.IOException;
import java.io.OutputStream;

@Tags({"custom", "processor"})
@CapabilityDescription("Custom processor example")
public class CustomProcessor extends AbstractProcessor {

    @Override
    public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
        session.read().attribute("myAttribute").stream()
                .forEach(inputStream -> {
                    try {
                        // Process the data and get the result
                        String result = processData(inputStream);

                        // Write the result to content
                        session.write(inputStream, new OutputStreamCallback() {
                            @Override
                            public void process(OutputStream outputStream) throws IOException {
                                outputStream.write(result.getBytes());
                            }
                        });

                        // Write the result to an attribute
                        session.putAttribute(inputStream, "result", result);

                        // Transfer the result to the next processor
                        session.transfer(inputStream, REL_SUCCESS);
                    } catch (Exception e) {
                        getLogger().error("Failed to process data", e);
                        session.transfer(inputStream, REL_FAILURE);
                    }
                });

        session.commit();
    }

    private String processData(InputStream inputStream) {
        // Custom data processing logic
        // ...

        return "Result";
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) {
        // Perform initialization tasks when the processor is scheduled to run
    }

    @OnUnscheduled
    public void onUnscheduled(final ProcessContext context) {
        // Perform cleanup tasks when the processor is unscheduled to run
    }

    @OnStopped
    public void onStopped(final ProcessContext context) {
        // Perform cleanup tasks when the processor is stopped
    }

    @OnShutdown
    public void onShutdown(final ProcessContext context) {
        // Perform cleanup tasks when the processor is shutdown
    }
}