Tboileau/testmultioutput

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

@@ -0,0 +1,14 @@
+package org.talend.sdk.component.api.processor;
+
+import java.util.Iterator;
+
+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.
+     *
+     * @param iterator the tagged iterator routing records to their named outputs
+     */
+    void setIterator(Iterator<TaggedOutput<T>> iterator);
+}

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

@@ -0,0 +1,61 @@
+/**
+ * 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 provide a lazy iterator for output records
+ * instead of pushing them via {@link OutputEmitter#emit(Object)}.
+ *
+ * <p>
+ * Used with {@code @Output} on an {@code OutputIterator} parameter to enable streaming:
+ * records are produced on-demand during the consumer's drain loop,
+ * rather than buffered entirely in memory.
+ *
+ * <p>
+ * <b>Important:</b> This interface is supported only in the Studio DI runtime.
+ * It is not supported in Beam-based runners (Cloud) where processors typically
+ * do not use TCK processor patterns. Using {@code OutputIterator} in a Beam
+ * pipeline will result in a {@code ClassCastException} at runtime.
+ *
+ * <p>
+ * Example usage in a processor:
+ * 
+ * <pre>
+ * {@code
+ * 
+ * &#64;ElementListener
+ * public void process(@Input Record input,
+ *         @Output OutputIterator<Record> output) {
+ *     output.setIterator(myLazyIterator(input));
+ * }
+ * }
+ * </pre>
+ *
+ * @param <T> the record type
+ */
+public interface OutputIterator<T> {
+
+    /**
+     * Sets the lazy iterator that will produce output records on demand.
+     * Call this once per invocation; the consumer will pull records
+     * by calling {@link Iterator#hasNext()} and {@link Iterator#next()}.
+     *
+     * @param iterator the lazy iterator providing output records
+     */
+    void setIterator(Iterator<T> iterator);
+}

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 rec;
+
+    /**
+     * Convenience factory method.
+     *
+     * @param outputName the target output connection name
+     * @param rec the record to emit
+     * @param <T> the record type
+     * @return a new TaggedOutput
+     */
+    public static <T> TaggedOutput<T> of(final String outputName, final T rec) {
+        return new TaggedOutput<>(outputName, rec);
+    }
+}

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

@@ -15,9 +15,41 @@
  */
 package org.talend.sdk.component.runtime.output;
 
+import org.talend.sdk.component.api.processor.MultiOutputIterator;
 import org.talend.sdk.component.api.processor.OutputEmitter;
+import org.talend.sdk.component.api.processor.OutputIterator;
 
 public interface OutputFactory {
 
     OutputEmitter create(String name);
+
+    /**
+     * Creates an {@link OutputIterator} for the given output branch name.
+     * Supported only in the Studio DI runtime. Other runtimes throw
+     * {@link UnsupportedOperationException} by default.
+     *
+     * @param name the output branch name
+     * @return an OutputIterator for lazy record streaming
+     * @throws UnsupportedOperationException if the runtime does not support iterator mode
+     */
+    default OutputIterator createIterator(String name) {
+        throw new UnsupportedOperationException(
+                "OutputIterator is only supported in the Studio DI runtime");
+    }
+
+    /**
+     * 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,7 +55,9 @@
 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.processor.OutputIterator;
 import org.talend.sdk.component.api.service.record.RecordBuilderFactory;
 import org.talend.sdk.component.runtime.base.Delegated;
 import org.talend.sdk.component.runtime.base.LifecycleImpl;
@@ -150,10 +152,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);
-            };
+            final String name = parameter.getAnnotation(Output.class).value();
+            if (OutputIterator.class == parameter.getType()) {
+                return (inputs, outputs) -> outputs.createIterator(name);
+            } else if (MultiOutputIterator.class == parameter.getType()) {
+                return (inputs, outputs) -> outputs.createMultiOutputIterator();
+            } else {
+                return (inputs, outputs) -> outputs.create(name);
+            }
         }
 
         final Class<?> parameterType = parameter.getType();
@@ -168,7 +174,13 @@ private Function<OutputFactory, Object> toOutputParamBuilder(final Parameter par
                 return false;
             }
             final String name = parameter.getAnnotation(Output.class).value();
-            return outputs.create(name);
+            if (OutputIterator.class == parameter.getType()) {
+                return outputs.createIterator(name);
+            } else if (MultiOutputIterator.class == parameter.getType()) {
+                return outputs.createMultiOutputIterator();
+            } else {
+                return outputs.create(name);
+            }
         };
     }
 

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

@@ -44,8 +44,10 @@
 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.OutputIterator;
 import org.talend.sdk.component.api.processor.Processor;
 import org.talend.sdk.component.api.standalone.DriverRunner;
 import org.talend.sdk.component.api.standalone.RunAtDriver;
@@ -195,7 +197,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 OutputIterator or MultiOutputIterator");
                 }
             })
                     .filter(p -> !p.isAnnotationPresent(Output.class))
@@ -243,7 +246,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 OutputIterator 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 +258,8 @@ private boolean validOutputParam(final Parameter p) {
         if (!(p.getParameterizedType() instanceof ParameterizedType pt)) {
             return false;
         }
-        return OutputEmitter.class == pt.getRawType();
+        final Type rawType = pt.getRawType();
+        return OutputEmitter.class == rawType || OutputIterator.class == rawType  || MultiOutputIterator.class == rawType;
     }
 
     private Stream<Class<? extends Annotation>> getPartitionMapperMethods(final boolean infinite) {

component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/BaseIOHandler.java

@@ -29,7 +29,9 @@
 import org.talend.sdk.component.runtime.record.RecordConverters;
 
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 public abstract class BaseIOHandler {
 
     protected final Jsonb jsonb;
@@ -84,26 +86,52 @@ protected String getActualName(final String name) {
         return "__default__".equals(name) ? "FLOW" : name;
     }
 
+    /**
+     * Represents a single output connection's data holder.
+     * Supports two modes (mutually exclusive per invocation):
+     * <ul>
+     * <li><b>Push mode</b> (default): records are added via {@link #add(Object)} into the internal queue.
+     * Used by {@code OutputEmitter.emit()}.</li>
+     * <li><b>Pull mode</b> (iterator): a lazy {@link Iterator} source is set via {@link #setSource(Iterator)}.
+     * Records are produced on-demand during the drain loop ({@link #hasNext()}/{@link #next()}).
+     * Used by {@code OutputIterator.setIterator()} in the Studio DI runtime.</li>
+     * </ul>
+     * Note: The {@code source} field is only used by {@code OutputsHandler}; {@code InputsHandler}
+     * uses only the queue-based push mode.
+     */
     @RequiredArgsConstructor
     static class IO<T> {
 
         private final Queue<T> values = new LinkedList<>();
 
         private final Class<T> type;
 
+        private Iterator<T> source;
+
+        void setSource(final Iterator<T> source) {
+            // Close previous source if it implements AutoCloseable
+            closeSource();
+            this.source = source;
+        }
+
         private void reset() {
             values.clear();
+            closeSource();
+            this.source = null;
         }
 
         boolean hasNext() {
-            return values.size() != 0;
+            return !values.isEmpty()
+                    || (source != null && source.hasNext());
         }
 
         T next() {
-            if (hasNext()) {
+            if (!values.isEmpty()) {
                 return type.cast(values.poll());
             }
-
+            if (source != null && source.hasNext()) {
+                return type.cast(source.next());
+            }
             return null;
         }
 
@@ -114,6 +142,16 @@ void add(final T e) {
         Class<T> getType() {
             return type;
         }
+
+        private void closeSource() {
+            if (source instanceof AutoCloseable) {
+                try {
+                    ((AutoCloseable) source).close();
+                } catch (final Exception e) {
+                    log.debug("Failed to close iterator source", e);
+                }
+            }
+        }
     }
 
 }