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,117 @@
+/**
+ * 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>
+ * <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 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 MultiOutputIterator<Record> out) {
+ *     out.setIterator("MAIN", mainDatabase.lazyQuery());
+ *     out.setIterator("REJECT", errorLog.lazyRead());
+ * }
+ * }
+ * </pre>
+ *
+ * <p>
+ * The drain loop on the Studio side uses {@code hasDataFor(name)} for correct
+ * per-connection checking in both modes:
+ *
+ * <pre>
+ * {@code
+ * while (outputsHandler.hasMoreData()) {
+ *     if (outputsHandler.hasDataFor("MAIN"))
+ *         mainRow = outputsHandler.getValue("MAIN");
+ *     if (outputsHandler.hasDataFor("REJECT"))
+ *         rejectRow = outputsHandler.getValue("REJECT");
+ * }
+ * }
+ * </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,5 @@
 @Retention(RUNTIME)
 public @interface Output {
 
-    String value() default "__default__";
+    String[] value() default { "__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

@@ -70,7 +70,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 -> of(p.getAnnotation(Output.class).value()));
     }
 
     private Stream<String> getReturnedBranches(final Method listener) {

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,11 @@ 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 = parameter.getAnnotation(Output.class).value()[0];
+            return (inputs, outputs) -> outputs.create(name);
         }
 
         final Class<?> parameterType = parameter.getType();
@@ -167,7 +169,10 @@ 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 = parameter.getAnnotation(Output.class).value()[0];
             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,8 @@ private boolean validOutputParam(final Parameter p) {
         if (!(p.getParameterizedType() instanceof ParameterizedType pt)) {
             return false;
         }
-        return OutputEmitter.class == pt.getRawType();
+        return OutputEmitter.class == pt.getRawType()
+                || MultiOutputIterator.class == pt.getRawType();
     }
 
     private Stream<Class<? extends Annotation>> getPartitionMapperMethods(final boolean infinite) {