diff --git a/component-api/src/main/java/org/talend/sdk/component/api/processor/MultiOutputIterator.java b/component-api/src/main/java/org/talend/sdk/component/api/processor/MultiOutputIterator.java new file mode 100644 index 0000000000000..ee6c8eed435f4 --- /dev/null +++ b/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 { + /** + * Split mode: 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> iterator); +} diff --git a/component-api/src/main/java/org/talend/sdk/component/api/processor/OutputIterator.java b/component-api/src/main/java/org/talend/sdk/component/api/processor/OutputIterator.java new file mode 100644 index 0000000000000..cfd62d32ebfe6 --- /dev/null +++ b/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)}. + * + *

+ * 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. + * + *

+ * Important: 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. + * + *

+ * Example usage in a processor: + * + * 

+ * {@code
+ * 
+ * @ElementListener
+ * public void process(@Input Record input,
+ *         @Output OutputIterator output) {
+ *     output.setIterator(myLazyIterator(input));
+ * }
+ * }
+ *
+ * + * @param the record type + */ +public interface OutputIterator { + + /** + * 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 iterator); +} diff --git a/component-api/src/main/java/org/talend/sdk/component/api/processor/TaggedOutput.java b/component-api/src/main/java/org/talend/sdk/component/api/processor/TaggedOutput.java new file mode 100644 index 0000000000000..c29da1bcfe65d --- /dev/null +++ b/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 the record type + */ +@Getter +@RequiredArgsConstructor +public class TaggedOutput { + + /** + * 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 the record type + * @return a new TaggedOutput + */ + public static TaggedOutput of(final String outputName, final T rec) { + return new TaggedOutput<>(outputName, rec); + } +} \ No newline at end of file diff --git a/component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/output/OutputFactory.java b/component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/output/OutputFactory.java index c3f8b14bf5fc1..cadb48ff172f8 100644 --- a/component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/output/OutputFactory.java +++ b/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. + * + *

+ * Supported only in the Studio DI runtime. + * + * @param the record type + * @return a MultiOutputIterator for lazy streaming + * @throws UnsupportedOperationException if the runtime does not support multi-output iterator mode + */ + default MultiOutputIterator createMultiOutputIterator() { + throw new UnsupportedOperationException( + "MultiOutputIterator is only supported in the Studio DI runtime"); + } } diff --git a/component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/output/ProcessorImpl.java b/component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/output/ProcessorImpl.java index 81229696456b1..fe7009347ed06 100644 --- a/component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/output/ProcessorImpl.java +++ b/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 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 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); + } }; } diff --git a/component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/visitor/ModelVisitor.java b/component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/visitor/ModelVisitor.java index 1b52f1b9ff546..91efcaaa8453d 100644 --- a/component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/visitor/ModelVisitor.java +++ b/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 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 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> getPartitionMapperMethods(final boolean infinite) { diff --git a/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/BaseIOHandler.java b/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/BaseIOHandler.java index 9b7441255171f..a25adff74677e 100644 --- a/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/BaseIOHandler.java +++ b/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,6 +86,19 @@ 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): + *

    + *
  • Push mode (default): records are added via {@link #add(Object)} into the internal queue. + * Used by {@code OutputEmitter.emit()}.
    • + *
  • Pull mode (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.
    • + *
+ * Note: The {@code source} field is only used by {@code OutputsHandler}; {@code InputsHandler} + * uses only the queue-based push mode. + */ @RequiredArgsConstructor static class IO { @@ -91,19 +106,32 @@ static class IO { private final Class type; + private Iterator source; + + void setSource(final Iterator 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 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); + } + } + } } } diff --git a/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/OutputsHandler.java b/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/OutputsHandler.java index 13bbee78e8ce9..2fec543a1829d 100644 --- a/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/OutputsHandler.java +++ b/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/OutputsHandler.java @@ -15,10 +15,15 @@ */ package org.talend.sdk.component.runtime.di; +import java.util.Iterator; import java.util.Map; import javax.json.bind.Jsonb; +import org.talend.sdk.component.api.processor.MultiOutputIterator; +import org.talend.sdk.component.api.processor.OutputEmitter; +import org.talend.sdk.component.api.processor.OutputIterator; +import org.talend.sdk.component.api.processor.TaggedOutput; import org.talend.sdk.component.api.record.Record; import org.talend.sdk.component.api.record.Schema; import org.talend.sdk.component.runtime.output.OutputFactory; @@ -33,20 +38,73 @@ public OutputsHandler(final Jsonb jsonb, final Map, Object> servicesMap } public OutputFactory asOutputFactory() { - return name -> value -> { - final BaseIOHandler.IO ref = connections.get(getActualName(name)); - if (ref != null && value != null) { - if (value instanceof javax.json.JsonValue) { - ref.add(jsonb.fromJson(value.toString(), ref.getType())); - } else if (value instanceof Record record) { - ref.add(registry.find(ref.getType()).newInstance(record)); - } else { - ref.add(jsonb.fromJson(jsonb.toJson(value), ref.getType())); - } + return new OutputFactory() { + + @Override + public OutputEmitter create(final String name) { + final BaseIOHandler.IO ref = connections.get(getActualName(name)); + return value -> { + if (ref != null && value != null) { + ref.add(convert(value, ref)); + } + }; + } + + @Override + public OutputIterator createIterator(final String name) { + final BaseIOHandler.IO ref = connections.get(getActualName(name)); + return iterator -> { // wrap the given iterator to provide converted values via the IO. + if (ref == null) { + return; + } + ref.setSource(new Iterator() { + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public Object next() { + return convert(iterator.next(), ref); + } + }); + }; + } + + @Override + public MultiOutputIterator createMultiOutputIterator() { + return topi -> setTaggedOutPutIterator(topi); } }; } + private void setTaggedOutPutIterator(final Iterator> taggedOutPutIterator) { + this.taggedOutPutIterator = taggedOutPutIterator; + } + + private Iterator> taggedOutPutIterator; + + @Override + public boolean hasMoreData() { + if (taggedOutPutIterator != null) { + if (taggedOutPutIterator.hasNext()) { + final TaggedOutput next = taggedOutPutIterator.next(); + final IO ref = connections.get(getActualName(next.getOutputName())); + final Object value = next.getRec(); + + if (ref != null && value != null) { + ref.add(convert(value, ref)); + } + return true; + } else { + return false; + } + } else { + return super.hasMoreData(); + } + } + /** * Guess schema special use-case for processor Studio mock. * Same as asOutputFactory but stores the record'schema or schema as the pojo class isn't available. @@ -59,8 +117,8 @@ public OutputFactory asOutputFactoryForGuessSchema() { if (ref != null && value != null) { if (value instanceof javax.json.JsonValue) { ref.add(jsonb.fromJson(value.toString(), ref.getType())); - } else if (value instanceof Record record) { - ref.add(record.getSchema()); + } else if (value instanceof Record rec) { + ref.add(rec.getSchema()); } else if (value instanceof Schema) { ref.add(value); } else { @@ -70,4 +128,15 @@ public OutputFactory asOutputFactoryForGuessSchema() { }; } + private Object convert(final Object value, final BaseIOHandler.IO ref) { + if (value == null) { + return null; + } else if (value instanceof javax.json.JsonValue) { + return jsonb.fromJson(value.toString(), ref.getType()); + } else if (value instanceof Record rec) { + return registry.find(ref.getType()).newInstance(rec); + } else { + return jsonb.fromJson(jsonb.toJson(value), ref.getType()); + } + } } diff --git a/component-studio/component-runtime-di/src/test/java/org/talend/sdk/component/runtime/di/studio/ProcessorBufferingTest.java b/component-studio/component-runtime-di/src/test/java/org/talend/sdk/component/runtime/di/studio/ProcessorBufferingTest.java new file mode 100644 index 0000000000000..0bf158f27c36b --- /dev/null +++ b/component-studio/component-runtime-di/src/test/java/org/talend/sdk/component/runtime/di/studio/ProcessorBufferingTest.java @@ -0,0 +1,465 @@ +/** + * 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.di.studio; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.File; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.HashMap; +import java.util.Map; +import java.util.stream.Stream; + +import javax.json.bind.Jsonb; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.talend.sdk.component.api.processor.AfterGroup; +import org.talend.sdk.component.api.processor.BeforeGroup; +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.Output; +import org.talend.sdk.component.api.processor.OutputIterator; +import org.talend.sdk.component.api.processor.Processor; +import org.talend.sdk.component.api.record.Record; +import org.talend.sdk.component.api.service.record.RecordBuilderFactory; +import org.talend.sdk.component.runtime.di.AutoChunkProcessor; +import org.talend.sdk.component.runtime.di.InputsHandler; +import org.talend.sdk.component.runtime.di.JobStateAware; +import org.talend.sdk.component.runtime.di.OutputsHandler; +import org.talend.sdk.component.runtime.manager.ComponentManager; +import org.talend.sdk.component.runtime.output.InputFactory; +import org.talend.sdk.component.runtime.output.OutputFactory; +import org.talend.sdk.component.runtime.record.RecordConverters; + +import lombok.Getter; +import lombok.ToString; + +/** + * Manual test to observe processor output buffering problem (QTDI-2709). + * + * Run this test to see memory consumption when a processor emits many records. + * With the current push model, ALL records are buffered in memory before the + * consumer can drain them. + * + * Expected behavior on current code: test FAILS (memory exceeds threshold). + * After iterator pattern is implemented: test PASSES (memory stays low). + */ +class ProcessorBufferingTest { + + protected static RecordBuilderFactory builderFactory; + + // Large enough to create observable memory pressure + static final int RECORD_COUNT = 100_000; + + // Memory threshold: if buffering all records uses more than this, the test fails. + // 100K records with strings should use >10MB when fully buffered. + // With streaming, memory should stay well under this. + static final long MAX_MEMORY_DELTA_MB = 5; + + enum OutputMode { + NO_OUTPUT, + ONE_OUTPUT, + TWO_OUTPUT + } + + @BeforeAll + static void forceManagerInit() { + final ComponentManager manager = ComponentManager.instance(); + if (manager.find(Stream::of).count() == 0) { + manager.addPlugin(new File("target/test-classes").getAbsolutePath()); + } + } + + /** + * Tests memory consumption when a processor emits 100K records in @ElementListener. + * + * FAILS on current code: all 100K records buffered → memory spike > threshold. + * PASSES after iterator implementation: records stream lazily → memory stays low. + */ + @ParameterizedTest + @EnumSource(OutputMode.class) + void elementListenerShouldNotBufferAllRecordsInMemory(OutputMode outputMode) { + final ComponentManager manager = ComponentManager.instance(); + final Map, Object> servicesMapper = getServicesMapper(manager); + final Jsonb jsonb = (Jsonb) servicesMapper.get(Jsonb.class); + builderFactory = (RecordBuilderFactory) servicesMapper.get(RecordBuilderFactory.class); + + final org.talend.sdk.component.runtime.output.Processor processor = manager + .findProcessor("ProcessorBufferingTest", "heavyEmitter", 1, new HashMap<>()) + .orElseThrow(() -> new IllegalStateException("processor not found")); + JobStateAware.init(processor, new HashMap<>()); + + final AutoChunkProcessor chunkProcessor = new AutoChunkProcessor(RECORD_COUNT + 1, processor); + chunkProcessor.start(); + + final InputsHandler inputsHandler = new InputsHandler(jsonb, servicesMapper); + inputsHandler.addConnection("FLOW", row1Struct.class); + + final OutputsHandler outputsHandler = new OutputsHandler(jsonb, servicesMapper); + if (outputMode != OutputMode.NO_OUTPUT) { + outputsHandler.addConnection("FLOW", row1Struct.class); + } + + final InputFactory inputFactory = inputsHandler.asInputFactory(); + final OutputFactory outputFactory = outputsHandler.asOutputFactory(); + + // Prepare one input record + final Record inputRecord = builderFactory.newRecordBuilder() + .withString("id", "input-1") + .withString("name", "trigger") + .build(); + final RecordConverters.MappingMetaRegistry registry = new RecordConverters.MappingMetaRegistry(); + final row1Struct inputRow = (row1Struct) registry.find(row1Struct.class).newInstance(inputRecord); + inputsHandler.setInputValue("FLOW", inputRow); + + // Force GC to get a clean baseline + gc(); + + final long memoryBefore = usedMemoryMB(); + + // --- Producer emits all records in onElement() --- + chunkProcessor.onElement(inputFactory, outputFactory); + + // Drain — same as Studio generated code (inside + outside loop, skipped for NO_OUTPUT) + int drainedCount = 0; + if (outputMode != OutputMode.NO_OUTPUT) { + // sim inside of input loop + while (outputsHandler.hasMoreData()) { + outputsHandler.getValue("FLOW"); + drainedCount++; + } + } + + chunkProcessor.flush(outputFactory); + + // GC to reclaim input record objects from the loop + gc(); + + // Measure memory AFTER production, BEFORE drain — valid for all output modes + final long memoryAfterProduction = usedMemoryMB(); + final long memoryDelta = memoryAfterProduction - memoryBefore; + + System.out.println("=== ProcessorBufferingTest: @ElementListener ==="); + System.out.println("Output mode: " + outputMode); + System.out.println("Records emitted: " + RECORD_COUNT); + System.out.println("Memory before onElement(): " + memoryBefore + " MB"); + System.out.println("Memory after onElement(): " + memoryAfterProduction + " MB"); + System.out.println("Memory delta: " + memoryDelta + " MB"); + System.out.println("Threshold: " + MAX_MEMORY_DELTA_MB + " MB"); + + if (outputMode != OutputMode.NO_OUTPUT) { + // sim outside of input loop + while (outputsHandler.hasMoreData()) { + outputsHandler.getValue("FLOW"); + drainedCount++; + } + } + System.out.println("Records drained: " + drainedCount); + + chunkProcessor.stop(); + + // ASSERTION: memory delta should be under threshold + // FAILS on current code (all records buffered → large delta) + // PASSES after iterator implementation (lazy streaming → small delta) + assertTrue(memoryDelta <= MAX_MEMORY_DELTA_MB, + "Memory delta should be <= " + MAX_MEMORY_DELTA_MB + " MB (streaming), " + + "but was " + memoryDelta + " MB (all records buffered in memory)"); + } + + /** + * Tests memory consumption when a processor emits 100K records in @AfterGroup. + * + * FAILS on current code: all records buffered in @AfterGroup. + * PASSES after iterator implementation. + */ + @ParameterizedTest + @EnumSource(OutputMode.class) + void afterGroupShouldNotBufferAllRecordsInMemory(OutputMode outputMode) { + final ComponentManager manager = ComponentManager.instance(); + final Map, Object> servicesMapper = getServicesMapper(manager); + final Jsonb jsonb = (Jsonb) servicesMapper.get(Jsonb.class); + builderFactory = (RecordBuilderFactory) servicesMapper.get(RecordBuilderFactory.class); + + final org.talend.sdk.component.runtime.output.Processor processor = manager + .findProcessor("ProcessorBufferingTest", "heavyAfterGroupEmitter", 1, new HashMap<>()) + .orElseThrow(() -> new IllegalStateException("processor not found")); + JobStateAware.init(processor, new HashMap<>()); + + // chunkSize=5: after 5 inputs, @AfterGroup fires and emits 100K records + final AutoChunkProcessor chunkProcessor = new AutoChunkProcessor(5, processor); + chunkProcessor.start(); + + final InputsHandler inputsHandler = new InputsHandler(jsonb, servicesMapper); + inputsHandler.addConnection("FLOW", row1Struct.class); + + final OutputsHandler outputsHandler = new OutputsHandler(jsonb, servicesMapper); + if (outputMode != OutputMode.NO_OUTPUT) { + outputsHandler.addConnection("MAIN", row1Struct.class); + if (outputMode == OutputMode.TWO_OUTPUT) { + outputsHandler.addConnection("REJECT", row1Struct.class); + } + } + + final InputFactory inputFactory = inputsHandler.asInputFactory(); + final OutputFactory outputFactory = outputsHandler.asOutputFactory(); + + final RecordConverters.MappingMetaRegistry registry = new RecordConverters.MappingMetaRegistry(); + + int mainCount = 0; + int rejectCount = 0; + + // Force GC to get a clean baseline + gc(); + final long memoryBefore = usedMemoryMB(); + + for (int i = 0; i < RECORD_COUNT; i++) { + final Record inputRecord = builderFactory.newRecordBuilder() + .withString("id", "input-" + i) + .withString("name", "record-" + i) + .build(); + final row1Struct inputRow = (row1Struct) registry.find(row1Struct.class).newInstance(inputRecord); + inputsHandler.setInputValue("FLOW", inputRow); + + chunkProcessor.onElement(inputFactory, outputFactory); + + if (outputMode != OutputMode.NO_OUTPUT) { + // Drain after each onElement — same as Studio generated code + while (outputsHandler.hasMoreData()) { + if (outputsHandler.getValue("MAIN") != null) { + mainCount++; + } + if (outputMode == OutputMode.TWO_OUTPUT) { + if (outputsHandler.getValue("REJECT") != null) { + rejectCount++; + } + } + } + } + } + + chunkProcessor.flush(outputFactory); + + // GC to reclaim input record objects from the loop + gc(); + + final long memoryAfterGroup = usedMemoryMB(); + final long memoryDelta = memoryAfterGroup - memoryBefore; + + System.out.println("=== ProcessorBufferingTest: @AfterGroup ==="); + System.out.println("Records emitted: " + RECORD_COUNT); + System.out.println("Memory before: " + memoryBefore + " MB"); + System.out.println("Memory after @AfterGroup: " + memoryAfterGroup + " MB"); + System.out.println("Memory delta: " + memoryDelta + " MB"); + + if (outputMode != OutputMode.NO_OUTPUT) { + while (outputsHandler.hasMoreData()) { + if (outputsHandler.getValue("MAIN") != null) { + mainCount++; + } + if (outputMode == OutputMode.TWO_OUTPUT) { + if (outputsHandler.getValue("REJECT") != null) { + rejectCount++; + } + } + } + + // Verify all records were produced + assertTrue(mainCount + rejectCount > 0, + "Should have produced records in @AfterGroup"); + } + + System.out.println("MAIN drained: " + mainCount); + System.out.println("REJECT drained: " + rejectCount); + + chunkProcessor.stop(); + + // ASSERTION: memory delta should be under threshold + // FAILS on current code (all records buffered → large delta) + // PASSES after iterator implementation (lazy streaming → small delta) + assertTrue(memoryDelta <= MAX_MEMORY_DELTA_MB, + "Memory delta should be <= " + MAX_MEMORY_DELTA_MB + " MB (streaming), " + + "but was " + memoryDelta + " MB (all records buffered in memory)"); + } + + // --- Helpers --- + + private static long usedMemoryMB() { + final Runtime rt = Runtime.getRuntime(); + return (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024); + } + + private Map, Object> getServicesMapper(ComponentManager manager) { + return manager + .findPlugin("test-classes") + .get() + .get(ComponentManager.AllServices.class) + .getServices(); + } + + // --- Test components --- + + /** + * Processor that provides a lazy iterator in @ElementListener. + * Records are produced on-demand during the drain loop. + */ + @Processor(name = "heavyEmitter", family = "ProcessorBufferingTest") + public static class HeavyEmitterProcessor implements Serializable { + + @ElementListener + public void onElement(@Input final Record input, + @Output final OutputIterator output) { + output.setIterator(new java.util.Iterator<>() { + + private int index = 0; + + @Override + public boolean hasNext() { + return index < RECORD_COUNT; + } + + @Override + public Record next() { + final Record rec = builderFactory.newRecordBuilder() + .withString("id", "out-" + index) + .withString("name", "generated-record-with-some-payload-" + index) + .withString("data", "additional-field-to-increase-memory-footprint-" + index) + .build(); + index++; + return rec; + } + }); + } + } + + /** + * Processor that accumulates inputs, then sets lazy iterators on MAIN and REJECT + * in @AfterGroup. Simulates the N:M batch pattern. + */ + @Processor(name = "heavyAfterGroupEmitter", family = "ProcessorBufferingTest") + public static class HeavyAfterGroupEmitterProcessor implements Serializable { + + private int inputCount = 0; + + @BeforeGroup + public void beforeGroup() { + // no-op + } + + @ElementListener + public void onElement(@Input final Record input) { + inputCount++; + } + + @AfterGroup + public void afterGroup(@Output("MAIN") final OutputIterator main, + @Output("REJECT") final OutputIterator reject, + @LastGroup final boolean lastGroup) { + if (!lastGroup) { + return; + } + + // MAIN gets records where index % 2 != 0 + main.setIterator(new java.util.Iterator<>() { + + private int index = 0; + + @Override + public boolean hasNext() { + while (index < RECORD_COUNT && index % 2 == 0) { + index++; + } + return index < RECORD_COUNT; + } + + @Override + public Record next() { + final Record rec = builderFactory.newRecordBuilder() + .withString("id", "result-" + index) + .withString("name", "processed-record-with-payload-" + index) + .withString("data", "bulk-result-data-field-" + index) + .build(); + index++; + return rec; + } + }); + + // REJECT gets records where index % 2 == 0 + reject.setIterator(new java.util.Iterator<>() { + + private int index = 0; + + @Override + public boolean hasNext() { + while (index < RECORD_COUNT && index % 2 != 0) { + index++; + } + return index < RECORD_COUNT; + } + + @Override + public Record next() { + final Record rec = builderFactory.newRecordBuilder() + .withString("id", "reject-" + index) + .withString("name", "rejected-record-" + index) + .withString("data", "reject-data-" + index) + .build(); + index++; + return rec; + } + }); + + inputCount = 0; + } + } + + // --- Row struct --- + + @Getter + @ToString + public static class row1Struct implements routines.system.IPersistableRow { + + public String id; + + public String name; + + public String data; + + @Override + public void writeData(final ObjectOutputStream objectOutputStream) { + throw new UnsupportedOperationException("#writeData()"); + } + + @Override + public void readData(final ObjectInputStream objectInputStream) { + throw new UnsupportedOperationException("#readData()"); + } + } + + private void gc() { + // GC to reclaim input record objects from the loop + System.gc(); + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +} diff --git a/component-tools/src/test/java/org/talend/sdk/component/tools/ComponentValidatorTest.java b/component-tools/src/test/java/org/talend/sdk/component/tools/ComponentValidatorTest.java index d4aa89522267a..65bdc0ea5e49b 100755 --- a/component-tools/src/test/java/org/talend/sdk/component/tools/ComponentValidatorTest.java +++ b/component-tools/src/test/java/org/talend/sdk/component/tools/ComponentValidatorTest.java @@ -571,7 +571,7 @@ void testFailureAfterGroup(final ExceptionSpec expectedException) { expectedException .expectMessage( """ - - @Output parameter must be of type OutputEmitter + - @Output parameter must be of type OutputEmitter or OutputIterator - Parameter of AfterGroup method need to be annotated with Output - class org.talend.test.failure.aftergroup.MyComponent5 must have a single @AfterGroup method with @LastGroup parameter"""); }