fix(QTDI-2709): TCK processors should send generated records to next connector without buffer

component-api/src/main/java/org/talend/sdk/component/api/processor/MultiOutputIterator.java

@@ -0,0 +1,108 @@
+/**
+ * Copyright (C) 2006-2026 Talend Inc. - www.talend.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.talend.sdk.component.api.processor;
+
+import java.util.Iterator;
+
+/**
+ * Allows a processor to stream records lazily to multiple output connections
+ * without buffering, supporting two mutually exclusive modes per invocation:
+ *
+ * <ul>
+ * <li><b>Split mode</b> ({@link #setIterator(Iterator)}): a single tagged iterator
+ * routes each record to a specific output connection via {@link TaggedOutput}.
+ * Use when records come from one shared source and must be split between outputs.</li>
+ * <li><b>Independent mode</b> ({@link #setIterator(String, Iterator)}): each output
+ * connection gets its own independent lazy iterator, consumed in parallel by the drain
+ * loop. Use when each output has its own self-contained lazy source.</li>
+ * </ul>
+ *
+ * <p>
+ * Both modes can be selected at runtime from the same fixed method parameter,
+ * so the component does not need separate parameters per output.
+ *
+ * <p>
+ * It is equally valid to use it with a <b>single</b> output: the point is then to stream records lazily
+ * instead of pushing them through an {@link OutputEmitter}, which avoids buffering. In that case the
+ * branch is named as usual with {@code @Output("FLOW")}; {@code @Output(branches = ...)} is only needed
+ * when the parameter feeds several branches, to declare them to the design layer.
+ *
+ * <p>
+ * <b>Important:</b> This interface is supported only in the Studio DI runtime.
+ *
+ * <p>
+ * <b>Split mode example</b> (one source, per-record routing):
+ *
+ * <pre>
+ * {@code
+ * 
+ * &#64;ElementListener
+ * public void process(&#64;Input Record input,
+ *         &#64;Output(branches = { "MAIN", "REJECT" }) MultiOutputIterator<Record> out) {
+ *     out.setIterator(
+ *             mySource.stream()
+ *                     .map(r -> isValid(r)
+ *                             ? TaggedOutput.of("MAIN", transform(r))
+ *                             : TaggedOutput.of("REJECT", r))
+ *                     .iterator());
+ * }
+ * }
+ * </pre>
+ *
+ * <p>
+ * <b>Independent mode example</b> (each output has its own lazy source):
+ *
+ * <pre>
+ * {@code
+ * 
+ * &#64;AfterGroup
+ * public void afterGroup(&#64;Output(branches = { "MAIN", "REJECT" }) MultiOutputIterator<Record> out) {
+ *     out.setIterator("MAIN", mainDatabase.lazyQuery());
+ *     out.setIterator("REJECT", errorLog.lazyRead());
+ * }
+ * }
+ * </pre>
+ *
+ * @param <T> the record type
+ * @see TaggedOutput
+ */
+public interface MultiOutputIterator<T> {
+
+    /**
+     * <b>Split mode</b>: sets a single lazy iterator whose elements are tagged with
+     * the target output connection name via {@link TaggedOutput}.
+     * The runtime reads one record at a time and routes it to the matching connection.
+     *
+     * <p>
+     * Mutually exclusive with {@link #setIterator(String, Iterator)} within one invocation.
+     *
+     * @param iterator the tagged iterator routing records to their named outputs
+     */
+    void setIterator(Iterator<TaggedOutput<T>> iterator);
+
+    /**
+     * <b>Independent mode</b>: assigns a lazy iterator to a specific named output connection.
+     * Call once per output that needs a dedicated lazy source; connections without an
+     * assigned iterator fall back to the push-mode queue as usual.
+     *
+     * <p>
+     * Mutually exclusive with {@link #setIterator(Iterator)} within one invocation.
+     *
+     * @param outputName the output connection name (e.g. {@code "MAIN"}, {@code "REJECT"})
+     * @param iterator the lazy iterator producing records for that connection
+     */
+    void setIterator(String outputName, Iterator<T> iterator);
+}

component-api/src/main/java/org/talend/sdk/component/api/processor/Output.java

@@ -25,5 +25,26 @@
 @Retention(RUNTIME)
 public @interface Output {
 
+    /**
+     * The output branch fed by this parameter.
+     *
+     * Used by {@link OutputEmitter} parameters, which always feed exactly one branch, and by
+     * {@link MultiOutputIterator} parameters streaming to a single branch.
+     *
+     * @return the branch name, {@code __default__} if not set.
+     */
     String value() default "__default__";
+
+    /**
+     * The branches a {@link MultiOutputIterator} parameter routes records to, when it feeds more than one.
+     *
+     * This is declarative metadata only: it lets the design/Studio layer know the output connections
+     * of the component, the routing itself is done at runtime through {@link TaggedOutput} or
+     * {@link MultiOutputIterator#setIterator(String, java.util.Iterator)}.
+     * When left empty, {@link #value()} is used, which covers the single branch streaming case.
+     * It is not supported on {@link OutputEmitter} parameters.
+     *
+     * @return the branch names, empty to fallback on {@link #value()}.
+     */
+    String[] branches() default {};
 }

component-api/src/main/java/org/talend/sdk/component/api/processor/TaggedOutput.java

@@ -0,0 +1,52 @@
+/**
+ * Copyright (C) 2006-2026 Talend Inc. - www.talend.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.talend.sdk.component.api.processor;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * A record tagged with its target output connection name.
+ * Used with {@link MultiOutputIterator} to route individual records to
+ * specific output connections from a single streaming iterator.
+ *
+ * @param <T> the record type
+ */
+@Getter
+@RequiredArgsConstructor
+public class TaggedOutput<T> {
+
+    /**
+     * The name of the output connection this record should be routed to.
+     * Use {@code "__default__"} or {@code "FLOW"} for the default output.
+     */
+    private final String outputName;
+
+    /** The record to emit to the named output. */
+    private final T record;
+
+    /**
+     * Convenience factory method.
+     *
+     * @param outputName the target output connection name
+     * @param record the record to emit
+     * @param <T> the record type
+     * @return a new TaggedOutput
+     */
+    public static <T> TaggedOutput<T> of(final String outputName, final T record) {
+        return new TaggedOutput<>(outputName, record);
+    }
+}

component-runtime-design-extension/src/main/java/org/talend/sdk/component/design/extension/flows/ProcessorFlowsFactory.java

@@ -30,6 +30,7 @@
 import org.talend.sdk.component.api.processor.Input;
 import org.talend.sdk.component.api.processor.Output;
 import org.talend.sdk.component.runtime.output.Branches;
+import org.talend.sdk.component.runtime.output.OutputBranches;
 
 import lombok.AllArgsConstructor;
 
@@ -70,7 +71,7 @@ private Optional<Method> getAfterGroup() {
     private Stream<String> getOutputParameters(final Method listener) {
         return of(listener.getParameters())
                 .filter(p -> p.isAnnotationPresent(Output.class))
-                .map(p -> p.getAnnotation(Output.class).value());
+                .flatMap(p -> OutputBranches.of(p.getAnnotation(Output.class)));
     }
 
     private Stream<String> getReturnedBranches(final Method listener) {

component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/output/OutputBranches.java

@@ -0,0 +1,42 @@
+/**
+ * Copyright (C) 2006-2026 Talend Inc. - www.talend.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.talend.sdk.component.runtime.output;
+
+import static lombok.AccessLevel.PRIVATE;
+
+import java.util.stream.Stream;
+
+import org.talend.sdk.component.api.processor.Output;
+
+import lombok.NoArgsConstructor;
+
+/**
+ * Utilities to read the branches declared by {@link Output} annotated parameters.
+ */
+@NoArgsConstructor(access = PRIVATE)
+public class OutputBranches {
+
+    /**
+     * Resolves the branches declared by an {@code @Output} parameter, whichever flavor was used:
+     * {@link Output#branches()} for a {@code MultiOutputIterator} parameter, {@link Output#value()} otherwise.
+     *
+     * @param output the annotation to read.
+     * @return the declared branch names.
+     */
+    public static Stream<String> of(final Output output) {
+        return output.branches().length > 0 ? Stream.of(output.branches()) : Stream.of(output.value());
+    }
+}

component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/output/OutputFactory.java

@@ -15,9 +15,26 @@
  */
 package org.talend.sdk.component.runtime.output;
 
+import org.talend.sdk.component.api.processor.MultiOutputIterator;
 import org.talend.sdk.component.api.processor.OutputEmitter;
 
 public interface OutputFactory {
 
     OutputEmitter create(String name);
+
+    /**
+     * Creates a {@link MultiOutputIterator} that routes records lazily to one or more
+     * output connections without buffering.
+     *
+     * <p>
+     * Supported only in the Studio DI runtime.
+     *
+     * @param <T> the record type
+     * @return a MultiOutputIterator for lazy streaming
+     * @throws UnsupportedOperationException if the runtime does not support multi-output iterator mode
+     */
+    default <T> MultiOutputIterator<T> createMultiOutputIterator() {
+        throw new UnsupportedOperationException(
+                "MultiOutputIterator is only supported in the Studio DI runtime");
+    }
 }

component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/output/ProcessorImpl.java

@@ -55,6 +55,7 @@
 import org.talend.sdk.component.api.processor.ElementListener;
 import org.talend.sdk.component.api.processor.Input;
 import org.talend.sdk.component.api.processor.LastGroup;
+import org.talend.sdk.component.api.processor.MultiOutputIterator;
 import org.talend.sdk.component.api.processor.Output;
 import org.talend.sdk.component.api.service.record.RecordBuilderFactory;
 import org.talend.sdk.component.runtime.base.Delegated;
@@ -150,10 +151,14 @@ public void beforeGroup() {
 
     private BiFunction<InputFactory, OutputFactory, Object> buildProcessParamBuilder(final Parameter parameter) {
         if (parameter.isAnnotationPresent(Output.class)) {
-            return (inputs, outputs) -> {
-                final String name = parameter.getAnnotation(Output.class).value();
-                return outputs.create(name);
-            };
+            if (MultiOutputIterator.class == parameter.getType()) {
+                return (inputs, outputs) -> outputs.createMultiOutputIterator();
+            }
+            final String name = OutputBranches
+                    .of(parameter.getAnnotation(Output.class))
+                    .findFirst()
+                    .orElse(Branches.DEFAULT_BRANCH);
+            return (inputs, outputs) -> outputs.create(name);
         }
 
         final Class<?> parameterType = parameter.getType();
@@ -167,7 +172,13 @@ private Function<OutputFactory, Object> toOutputParamBuilder(final Parameter par
             if (parameter.isAnnotationPresent(LastGroup.class)) {
                 return false;
             }
-            final String name = parameter.getAnnotation(Output.class).value();
+            if (MultiOutputIterator.class == parameter.getType()) {
+                return outputs.createMultiOutputIterator();
+            }
+            final String name = OutputBranches
+                    .of(parameter.getAnnotation(Output.class))
+                    .findFirst()
+                    .orElse(Branches.DEFAULT_BRANCH);
             return outputs.create(name);
         };
     }

component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/visitor/ModelVisitor.java

@@ -44,6 +44,7 @@
 import org.talend.sdk.component.api.processor.BeforeGroup;
 import org.talend.sdk.component.api.processor.ElementListener;
 import org.talend.sdk.component.api.processor.LastGroup;
+import org.talend.sdk.component.api.processor.MultiOutputIterator;
 import org.talend.sdk.component.api.processor.Output;
 import org.talend.sdk.component.api.processor.OutputEmitter;
 import org.talend.sdk.component.api.processor.Processor;
@@ -195,7 +196,8 @@ private void validateProcessor(final Class<?> input) {
         afterGroups.forEach(m -> {
             final List<Parameter> invalidParams = Stream.of(m.getParameters()).peek(p -> {
                 if (p.isAnnotationPresent(Output.class) && !validOutputParam(p)) {
-                    throw new IllegalArgumentException("@Output parameter must be of type OutputEmitter");
+                    throw new IllegalArgumentException(
+                            "@Output parameter must be of type OutputEmitter or MultiOutputIterator");
                 }
             })
                     .filter(p -> !p.isAnnotationPresent(Output.class))
@@ -243,7 +245,8 @@ private void validateProducer(final Class<?> input, final List<Method> afterGrou
 
         if (!producers.isEmpty() && Stream.of(producers.get(0).getParameters()).peek(p -> {
             if (p.isAnnotationPresent(Output.class) && !validOutputParam(p)) {
-                throw new IllegalArgumentException("@Output parameter must be of type OutputEmitter");
+                throw new IllegalArgumentException(
+                        "@Output parameter must be of type OutputEmitter or MultiOutputIterator");
             }
         }).filter(p -> !p.isAnnotationPresent(Output.class)).count() < 1) {
             throw new IllegalArgumentException(input + " doesn't have the input parameter on its producer method");
@@ -254,7 +257,14 @@ private boolean validOutputParam(final Parameter p) {
         if (!(p.getParameterizedType() instanceof ParameterizedType pt)) {
             return false;
         }
-        return OutputEmitter.class == pt.getRawType();
+        if (OutputEmitter.class == pt.getRawType()) {
+            if (p.getAnnotation(Output.class).branches().length > 0) {
+                throw new IllegalArgumentException(
+                        "@Output#branches is only supported on MultiOutputIterator parameters, use @Output(\"name\")");
+            }
+            return true;
+        }
+        return MultiOutputIterator.class == pt.getRawType();
     }
 
     private Stream<Class<? extends Annotation>> getPartitionMapperMethods(final boolean infinite) {