From 01e44a21f92dfe2e2bfd8a63ce3c61467e8499c9 Mon Sep 17 00:00:00 2001 From: Jin Zhao Date: Fri, 14 Aug 2026 15:37:07 +0800 Subject: [PATCH] =?UTF-8?q?Revert=20"fix(QTDI-2709):=20TCK=20processors=20?= =?UTF-8?q?should=20send=20generated=20records=20to=20next=20=E2=80=A6"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit 31fc9739ba3e766615ec114f865d79554761b180. --- .../api/processor/MultiOutputIterator.java | 102 -- .../sdk/component/api/processor/Output.java | 2 +- .../component/api/processor/TaggedOutput.java | 52 - .../flows/ProcessorFlowsFactory.java | 2 +- .../runtime/output/OutputFactory.java | 17 - .../runtime/output/ProcessorImpl.java | 15 +- .../runtime/visitor/ModelVisitor.java | 10 +- .../component/runtime/di/BaseIOHandler.java | 184 +--- .../component/runtime/di/OutputsHandler.java | 81 +- .../runtime/di/schema/TaCoKitGuessSchema.java | 7 +- .../di/studio/ProcessorBufferingTest.java | 967 ------------------ .../tools/ComponentValidatorTest.java | 2 +- 12 files changed, 30 insertions(+), 1411 deletions(-) delete mode 100644 component-api/src/main/java/org/talend/sdk/component/api/processor/MultiOutputIterator.java delete mode 100644 component-api/src/main/java/org/talend/sdk/component/api/processor/TaggedOutput.java delete mode 100644 component-studio/component-runtime-di/src/test/java/org/talend/sdk/component/runtime/di/studio/ProcessorBufferingTest.java 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 deleted file mode 100644 index 9d08d30db73ee..0000000000000 --- a/component-api/src/main/java/org/talend/sdk/component/api/processor/MultiOutputIterator.java +++ /dev/null @@ -1,102 +0,0 @@ -/** - * 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: - * - * - * - *

- * Both modes can be selected at runtime from the same fixed method parameter, - * so the component does not need separate parameters per output. - * - *

- * Important: This interface is supported only in the Studio DI runtime. - * - *

- * Split mode example (one source, per-record routing): - * - * 

- * {@code
- * 
- * @ElementListener
- * public void process(@Input Record input,
- *         @Output MultiOutputIterator out) {
- *     out.setIterator(
- *             mySource.stream()
- *                     .map(r -> isValid(r)
- *                             ? TaggedOutput.of("MAIN", transform(r))
- *                             : TaggedOutput.of("REJECT", r))
- *                     .iterator());
- * }
- * }
- *
- * - *

- * Independent mode example (each output has its own lazy source): - * - * 

- * {@code
- * 
- * @AfterGroup
- * public void afterGroup(@Output MultiOutputIterator out) {
- *     out.setIterator("MAIN", mainDatabase.lazyQuery());
- *     out.setIterator("REJECT", errorLog.lazyRead());
- * }
- * }
- *
- * - * @param the record type - * @see TaggedOutput - */ -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. - * - *

- * Mutually exclusive with {@link #setIterator(String, Iterator)} within one invocation. - * - * @param iterator the tagged iterator routing records to their named outputs - */ - void setIterator(Iterator> iterator); - - /** - * Independent mode: 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. - * - *

- * 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 iterator); -} diff --git a/component-api/src/main/java/org/talend/sdk/component/api/processor/Output.java b/component-api/src/main/java/org/talend/sdk/component/api/processor/Output.java index 6d1ea3a7ae3f5..a6fde9f1e70b2 100644 --- a/component-api/src/main/java/org/talend/sdk/component/api/processor/Output.java +++ b/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__"; } 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 deleted file mode 100644 index a6bf6fb5c1d80..0000000000000 --- a/component-api/src/main/java/org/talend/sdk/component/api/processor/TaggedOutput.java +++ /dev/null @@ -1,52 +0,0 @@ -/** - * 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 record; - - /** - * Convenience factory method. - * - * @param outputName the target output connection name - * @param record the record to emit - * @param the record type - * @return a new TaggedOutput - */ - public static TaggedOutput of(final String outputName, final T record) { - return new TaggedOutput<>(outputName, record); - } -} diff --git a/component-runtime-design-extension/src/main/java/org/talend/sdk/component/design/extension/flows/ProcessorFlowsFactory.java b/component-runtime-design-extension/src/main/java/org/talend/sdk/component/design/extension/flows/ProcessorFlowsFactory.java index bff51ecf522a3..37ddce11cf0b8 100644 --- a/component-runtime-design-extension/src/main/java/org/talend/sdk/component/design/extension/flows/ProcessorFlowsFactory.java +++ b/component-runtime-design-extension/src/main/java/org/talend/sdk/component/design/extension/flows/ProcessorFlowsFactory.java @@ -70,7 +70,7 @@ private Optional getAfterGroup() { private Stream getOutputParameters(final Method listener) { return of(listener.getParameters()) .filter(p -> p.isAnnotationPresent(Output.class)) - .flatMap(p -> of(p.getAnnotation(Output.class).value())); + .map(p -> p.getAnnotation(Output.class).value()); } private Stream getReturnedBranches(final Method listener) { 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 44d09b04c78ce..c3f8b14bf5fc1 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,26 +15,9 @@ */ 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. - * - *

- * 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 fa08b2a7846e1..81229696456b1 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,6 @@ 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; @@ -151,11 +150,10 @@ public void beforeGroup() { private BiFunction buildProcessParamBuilder(final Parameter parameter) { if (parameter.isAnnotationPresent(Output.class)) { - 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); + return (inputs, outputs) -> { + final String name = parameter.getAnnotation(Output.class).value(); + return outputs.create(name); + }; } final Class parameterType = parameter.getType(); @@ -169,10 +167,7 @@ private Function toOutputParamBuilder(final Parameter par if (parameter.isAnnotationPresent(LastGroup.class)) { return false; } - if (MultiOutputIterator.class == parameter.getType()) { - return outputs.createMultiOutputIterator(); - } - final String name = parameter.getAnnotation(Output.class).value()[0]; + final String name = parameter.getAnnotation(Output.class).value(); 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 f34e928f28702..1b52f1b9ff546 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,7 +44,6 @@ 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; @@ -196,8 +195,7 @@ 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 or MultiOutputIterator"); + throw new IllegalArgumentException("@Output parameter must be of type OutputEmitter"); } }) .filter(p -> !p.isAnnotationPresent(Output.class)) @@ -245,8 +243,7 @@ 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 or MultiOutputIterator"); + throw new IllegalArgumentException("@Output parameter must be of type OutputEmitter"); } }).filter(p -> !p.isAnnotationPresent(Output.class)).count() < 1) { throw new IllegalArgumentException(input + " doesn't have the input parameter on its producer method"); @@ -257,8 +254,7 @@ private boolean validOutputParam(final Parameter p) { if (!(p.getParameterizedType() instanceof ParameterizedType pt)) { return false; } - return OutputEmitter.class == pt.getRawType() - || MultiOutputIterator.class == pt.getRawType(); + return OutputEmitter.class == pt.getRawType(); } 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 136e3c7a4532b..9b7441255171f 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,9 +29,7 @@ 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; @@ -42,31 +40,6 @@ public abstract class BaseIOHandler { protected final Map connections = new TreeMap<>(); - /** - * Functional interface used internally to advance the tagged source one step. - * Implementations call {@link #setPending(String, Object)} to store the next - * record's output name and converted value, then return {@code true}. - * Return {@code false} when the source is exhausted. - */ - @FunctionalInterface - protected interface TaggedAdvancer { - - boolean advance(); - } - - /** - * Tagged-source advancer for split streaming. - * Non-null only when the component used a - * {@link org.talend.sdk.component.api.processor.MultiOutputIterator}. - */ - private TaggedAdvancer taggedSource; - - /** Output-connection name of the look-ahead record; {@code null} when no record is pending. */ - private String pendingOutputName; - - /** Converted record value paired with {@link #pendingOutputName}. */ - private Object pendingRecord; - public BaseIOHandler(final Jsonb jsonb, final Map, Object> servicesMapper) { this.jsonb = jsonb; this.recordBuilderMapper = (RecordBuilderFactory) servicesMapper.get(RecordBuilderFactory.class); @@ -96,144 +69,21 @@ public void addConnection(final String connectorName, final Class type) { } public void reset() { - for (IO value : connections.values()) { - value.reset(); - } - taggedSource = null; - pendingOutputName = null; - pendingRecord = null; + connections.values().forEach(IO::reset); } public T getValue(final String connectorName) { - if (taggedSource != null) { - if (connectorName.equals(pendingOutputName)) { - final T value = (T) pendingRecord; - pendingOutputName = null; - pendingRecord = null; - return value; - } - return null; - } return (T) connections.get(connectorName).next(); } public boolean hasMoreData() { - if (taggedSource != null) { - return pendingOutputName != null || taggedSource.advance(); - } - for (IO value : connections.values()) { - if (value.hasNext()) { - return true; - } - } - return false; - } - - /** - * Returns {@code true} if the named connection has a record ready to be consumed. - * - *

- * In tagged-source mode (when the component used a - * {@link org.talend.sdk.component.api.processor.MultiOutputIterator}), - * this peeks one record ahead from the shared tagged source and returns whether it - * belongs to {@code connectionName}. The peeked record is buffered and consumed by - * the next {@link #getValue(String)} call for the same connection. - * - *

- * In independent-iterator mode, this directly checks the per-connection iterator/queue. - * - *

- * Use this method in the Studio drain loop to avoid calling {@link #getValue(String)} - * on connections that have no data: - * - * 

{@code
-     * while (outputsHandler.hasMoreData()) {
-     *     if (outputsHandler.hasDataFor("MAIN"))
-     *         mainRow = outputsHandler.getValue("MAIN");
-     *     if (outputsHandler.hasDataFor("REJECT"))
-     *         rejectRow = outputsHandler.getValue("REJECT");
-     * }
-     * }
- * - * @param connectionName the output connection name to check - * @return {@code true} if a record for this connection is immediately available - */ - public boolean hasDataFor(final String connectionName) { - if (taggedSource != null) { - // Skip any pending record for an unregistered connection before checking - while (pendingOutputName != null && !connections.containsKey(pendingOutputName)) { - pendingOutputName = null; - pendingRecord = null; - if (!taggedSource.advance()) { - return false; - } - } - if (pendingOutputName != null) { - return pendingOutputName.equals(connectionName); - } - // Advance, skipping records for unregistered connections - while (taggedSource.advance()) { - if (connections.containsKey(pendingOutputName)) { - return pendingOutputName.equals(connectionName); - } - pendingOutputName = null; - pendingRecord = null; - } - return false; - } - final IO io = connections.get(connectionName); - return io != null && io.hasNext(); - } - - /** - * Sets a tagged-source advancer for split streaming across multiple outputs. - * Switching to tagged mode — subsequent {@link #hasMoreData()}, {@link #hasDataFor(String)}, - * and {@link #getValue(String)} operate via the advancer instead of per-connection queues. - * - *

- * The {@code advancer} implementation must call {@link #setPending(String, Object)} with - * the next output-connection name and converted record value, then return {@code true}. - * Return {@code false} when the source is exhausted. - * - * @param advancer the tagged-source advancer produced by a MultiOutputIterator setup - */ - protected void setTaggedSource(final TaggedAdvancer advancer) { - this.taggedSource = advancer; - this.pendingOutputName = null; - this.pendingRecord = null; - } - - /** - * Called by a {@link TaggedAdvancer} to store the next pending record. - * - * @param outputName the output connection name for the record - * @param record the converted record value - */ - protected void setPending(final String outputName, final Object record) { - this.pendingOutputName = outputName; - this.pendingRecord = record; + return connections.entrySet().stream().anyMatch(e -> e.getValue().hasNext()); } 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 MultiOutputIterator.setIterator(String, Iterator)} in the Studio DI runtime.
    • - *
- * In pull mode, {@link #hasNext()} delegates directly to the source iterator's - * {@code hasNext()} — the source iterator is expected to be idempotent and fast. - *

- * Note: The {@code source} field is only used by {@code OutputsHandler}; {@code InputsHandler} - * uses only the queue-based push mode. - */ @RequiredArgsConstructor static class IO { @@ -241,33 +91,19 @@ static class IO { private final Class type; - private Iterator source; - - void setSource(final Iterator source) { - closeSource(); - this.source = source; - } - private void reset() { values.clear(); - closeSource(); - this.source = null; } boolean hasNext() { - if (!values.isEmpty()) { - return true; - } - return source != null && source.hasNext(); + return values.size() != 0; } T next() { - if (!values.isEmpty()) { + if (hasNext()) { return type.cast(values.poll()); } - if (source != null && source.hasNext()) { - return type.cast(source.next()); - } + return null; } @@ -278,16 +114,6 @@ 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 b8de1ad38ee4e..13bbee78e8ce9 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,14 +15,10 @@ */ 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.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; @@ -37,57 +33,16 @@ public OutputsHandler(final Jsonb jsonb, final Map, Object> servicesMap } public OutputFactory asOutputFactory() { - 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 MultiOutputIterator createMultiOutputIterator() { - return new MultiOutputIterator() { - - @Override - public void setIterator(final Iterator> iterator) { - setTaggedSource(() -> { - if (!iterator.hasNext()) { - return false; - } - final TaggedOutput tagged = iterator.next(); - final String name = tagged.getOutputName(); - final BaseIOHandler.IO ref = connections.get(name); - setPending(name, - ref != null ? convert(tagged.getRecord(), ref) : tagged.getRecord()); - return true; - }); - } - - @Override - public void setIterator(final String outputName, final Iterator iterator) { - final BaseIOHandler.IO ref = connections.get(outputName); - if (ref == null) { - return; - } - ref.setSource(new Iterator() { - - @Override - public boolean hasNext() { - return iterator.hasNext(); - } - - @Override - public Object next() { - return convert(iterator.next(), ref); - } - }); - } - }; + 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())); + } } }; } @@ -104,8 +59,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 rec) { - ref.add(rec.getSchema()); + } else if (value instanceof Record record) { + ref.add(record.getSchema()); } else if (value instanceof Schema) { ref.add(value); } else { @@ -115,16 +70,4 @@ 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/main/java/org/talend/sdk/component/runtime/di/schema/TaCoKitGuessSchema.java b/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/schema/TaCoKitGuessSchema.java index ae18b78bdca41..7dad572ad4202 100644 --- a/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/schema/TaCoKitGuessSchema.java +++ b/component-studio/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/schema/TaCoKitGuessSchema.java @@ -53,7 +53,6 @@ import org.talend.sdk.component.api.exception.ComponentException; import org.talend.sdk.component.api.exception.DiscoverSchemaException; import org.talend.sdk.component.api.processor.ElementListener; -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.record.Record; @@ -777,12 +776,10 @@ public void fromOutputEmitterPojo(final Processor processor, final String outBra .flatMap(m -> IntStream .range(0, m.getParameterCount()) .filter(i -> m.getParameters()[i].isAnnotationPresent(Output.class) - && java.util.Arrays.asList(m.getParameters()[i].getAnnotation(Output.class).value()) - .contains(outBranchName)) + && outBranchName.equals(m.getParameters()[i].getAnnotation(Output.class).value())) .mapToObj(i -> m.getGenericParameterTypes()[i]) .filter(t -> t instanceof ParameterizedType parameterizedType - && (parameterizedType.getRawType() == OutputEmitter.class - || parameterizedType.getRawType() == MultiOutputIterator.class) + && parameterizedType.getRawType() == OutputEmitter.class && parameterizedType.getActualTypeArguments().length == 1) .map(p -> ((ParameterizedType) p).getActualTypeArguments()[0])) .findFirst(); 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 deleted file mode 100644 index 53c426d10b160..0000000000000 --- a/component-studio/component-runtime-di/src/test/java/org/talend/sdk/component/runtime/di/studio/ProcessorBufferingTest.java +++ /dev/null @@ -1,967 +0,0 @@ -/** - * 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 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.Assertions; -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.MultiOutputIterator; -import org.talend.sdk.component.api.processor.Output; -import org.talend.sdk.component.api.processor.Processor; -import org.talend.sdk.component.api.processor.TaggedOutput; -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()) { - final row1Struct rec = outputsHandler.getValue("FLOW"); - Assertions.assertNotNull(rec, "FLOW record should not be null"); - Assertions.assertEquals("out-" + drainedCount, rec.id, - "FLOW record id mismatch at index " + drainedCount); - 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()) { - final row1Struct rec = outputsHandler.getValue("FLOW"); - Assertions.assertNotNull(rec, "FLOW record should not be null"); - Assertions.assertEquals("out-" + drainedCount, rec.id, - "FLOW record id mismatch at index " + drainedCount); - 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) - Assertions.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()) { - final row1Struct mainRow = outputsHandler.getValue("MAIN"); - if (mainRow != null) { - Assertions.assertEquals("result-" + (mainCount * 2 + 1), mainRow.id, - "MAIN record id mismatch at index " + mainCount); - mainCount++; - } - if (outputMode == OutputMode.TWO_OUTPUT) { - final row1Struct rejectRow = outputsHandler.getValue("REJECT"); - if (rejectRow != null) { - Assertions.assertEquals("reject-" + (rejectCount * 2), rejectRow.id, - "REJECT record id mismatch at index " + rejectCount); - 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()) { - final row1Struct mainRow = outputsHandler.getValue("MAIN"); - if (mainRow != null) { - Assertions.assertEquals("result-" + (mainCount * 2 + 1), mainRow.id, - "MAIN record id mismatch at index " + mainCount); - mainCount++; - } - if (outputMode == OutputMode.TWO_OUTPUT) { - final row1Struct rejectRow = outputsHandler.getValue("REJECT"); - if (rejectRow != null) { - Assertions.assertEquals("reject-" + (rejectCount * 2), rejectRow.id, - "REJECT record id mismatch at index " + rejectCount); - rejectCount++; - } - } - } - - // Verify all records were produced - Assertions.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) - Assertions.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 that a processor using a split {@link MultiOutputIterator} in {@code @AfterGroup} - * correctly routes records between MAIN and REJECT via {@code hasDataFor} without buffering. - * - *

- * Uses a single shared tagged iterator that routes even-indexed records to MAIN and - * odd-indexed records to REJECT. The drain loop uses {@code hasDataFor} to consume - * only the connection that has a record ready — which exercises the fix in - * {@code hasDataFor} that skips unregistered pending records instead of stalling. - */ - @ParameterizedTest - @EnumSource(OutputMode.class) - void afterGroupSplitIteratorShouldRouteRecordsWithHasDataFor(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", "splitAfterGroupEmitter", 1, new HashMap<>()) - .orElseThrow(() -> new IllegalStateException("splitAfterGroupEmitter processor not found")); - JobStateAware.init(processor, new HashMap<>()); - - // chunkSize=5: after 5 inputs, @AfterGroup fires and emits 100K records via split iterator - 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; - - gc(); - final long memoryBefore = usedMemoryMB(); - - for (int i = 0; i < 5; 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 using hasDataFor — exercises the fix that skips unregistered pending records - while (outputsHandler.hasMoreData()) { - if (outputsHandler.hasDataFor("MAIN")) { - final row1Struct mainRow = outputsHandler.getValue("MAIN"); - Assertions.assertNotNull(mainRow, "MAIN record should not be null"); - Assertions.assertEquals("main-" + (mainCount * 2), mainRow.id, - "MAIN record id mismatch at index " + mainCount); - mainCount++; - } - if (outputMode == OutputMode.TWO_OUTPUT && outputsHandler.hasDataFor("REJECT")) { - final row1Struct rejectRow = outputsHandler.getValue("REJECT"); - Assertions.assertNotNull(rejectRow, "REJECT record should not be null"); - Assertions.assertEquals("reject-" + (rejectCount * 2 + 1), rejectRow.id, - "REJECT record id mismatch at index " + rejectCount); - rejectCount++; - } - } - } - } - - chunkProcessor.flush(outputFactory); - - // Drain after flush — @AfterGroup with lastGroup=true fires inside flush(), - // so the split iterator is only set there and must be drained here. - if (outputMode != OutputMode.NO_OUTPUT) { - while (outputsHandler.hasMoreData()) { - if (outputsHandler.hasDataFor("MAIN")) { - final row1Struct mainRow = outputsHandler.getValue("MAIN"); - Assertions.assertNotNull(mainRow, "MAIN record should not be null"); - Assertions.assertEquals("main-" + (mainCount * 2), mainRow.id, - "MAIN record id mismatch at index " + mainCount); - mainCount++; - } - if (outputMode == OutputMode.TWO_OUTPUT && outputsHandler.hasDataFor("REJECT")) { - final row1Struct rejectRow = outputsHandler.getValue("REJECT"); - Assertions.assertNotNull(rejectRow, "REJECT record should not be null"); - Assertions.assertEquals("reject-" + (rejectCount * 2 + 1), rejectRow.id, - "REJECT record id mismatch at index " + rejectCount); - rejectCount++; - } - } - } - - gc(); - - final long memoryAfter = usedMemoryMB(); - final long memoryDelta = memoryAfter - memoryBefore; - - System.out.println("=== ProcessorBufferingTest: @AfterGroup split iterator ==="); - System.out.println("Output mode: " + outputMode); - System.out.println("MAIN drained: " + mainCount); - System.out.println("REJECT drained: " + rejectCount); - System.out.println("Memory delta: " + memoryDelta + " MB"); - - final int half = RECORD_COUNT / 2; - switch (outputMode) { - case TWO_OUTPUT: - Assertions.assertEquals(half, mainCount, - "MAIN should receive half the records (even indices)"); - Assertions.assertEquals(half, rejectCount, - "REJECT should receive half the records (odd indices)"); - break; - case ONE_OUTPUT: - Assertions.assertEquals(half, mainCount, - "MAIN should receive half the records"); - Assertions.assertEquals(0, rejectCount, - "REJECT not registered, should receive nothing"); - break; - case NO_OUTPUT: - Assertions.assertEquals(0, mainCount + rejectCount, - "No connections registered, no records should be received"); - break; - } - Assertions.assertTrue(memoryDelta <= MAX_MEMORY_DELTA_MB, - "Memory delta should be <= " + MAX_MEMORY_DELTA_MB + " MB, was " + memoryDelta + " MB"); - - chunkProcessor.stop(); - } - - // --- 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(); - } - - /** - * Tests that a processor using {@link MultiOutputIterator} correctly splits records - * between MAIN and REJECT without buffering and without requiring lockstep iteration. - * - *

- * The split processor routes even-indexed records to MAIN and odd-indexed to REJECT - * from a SINGLE shared source iterator. Without tagged-output support this would - * require buffering all records first. - */ - @ParameterizedTest - @EnumSource(OutputMode.class) - void multiOutputIteratorShouldSplitRecordsWithoutBuffering(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", "splitEmitter", 1, new HashMap<>()) - .orElseThrow(() -> new IllegalStateException("splitEmitter 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("MAIN", row1Struct.class); - if (outputMode == OutputMode.TWO_OUTPUT) { - outputsHandler.addConnection("REJECT", row1Struct.class); - } - } - - final InputFactory inputFactory = inputsHandler.asInputFactory(); - final OutputFactory outputFactory = outputsHandler.asOutputFactory(); - - 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); - - gc(); - final long memoryBefore = usedMemoryMB(); - - chunkProcessor.onElement(inputFactory, outputFactory); - - // Drain using per-connection hasDataFor checks — correct for split streaming - int mainCount = 0; - int rejectCount = 0; - if (outputMode != OutputMode.NO_OUTPUT) { - while (outputsHandler.hasMoreData()) { - if (outputsHandler.hasDataFor("MAIN")) { - final row1Struct mainRow = outputsHandler.getValue("MAIN"); - Assertions.assertNotNull(mainRow, "MAIN record should not be null"); - Assertions.assertEquals("main-" + (mainCount * 2), mainRow.id, - "MAIN record id mismatch at index " + mainCount); - mainCount++; - } - if (outputMode == OutputMode.TWO_OUTPUT) { - if (outputsHandler.hasDataFor("REJECT")) { - final row1Struct rejectRow = outputsHandler.getValue("REJECT"); - Assertions.assertNotNull(rejectRow, "REJECT record should not be null"); - Assertions.assertEquals("reject-" + (rejectCount * 2 + 1), rejectRow.id, - "REJECT record id mismatch at index " + rejectCount); - rejectCount++; - } - } - } - } - - chunkProcessor.flush(outputFactory); - gc(); - final long memoryDelta = usedMemoryMB() - memoryBefore; - - System.out.println("=== ProcessorBufferingTest: MultiOutputIterator split ==="); - System.out.println("Records emitted: " + RECORD_COUNT); - System.out.println("MAIN drained: " + mainCount); - System.out.println("REJECT drained: " + rejectCount); - System.out.println("Memory delta: " + memoryDelta + " MB"); - - final int expected = RECORD_COUNT / 2; - switch (outputMode) { - case TWO_OUTPUT: - // Both connections registered: each receives exactly half - Assertions.assertEquals(expected, mainCount, - "MAIN should receive half the records"); - Assertions.assertEquals(expected, rejectCount, - "REJECT should receive the other half"); - break; - case ONE_OUTPUT: - // Only MAIN registered: MAIN gets its half, REJECT records are skipped - Assertions.assertEquals(expected, mainCount, - "MAIN should receive half the records"); - Assertions.assertEquals(0, rejectCount, - "REJECT not registered, should receive nothing"); - break; - case NO_OUTPUT: - // No connections registered: nothing drained - Assertions.assertEquals(0, mainCount + rejectCount, - "No connections registered, no records should be received"); - break; - } - Assertions.assertTrue(memoryDelta <= MAX_MEMORY_DELTA_MB, - "Memory delta should be <= " + MAX_MEMORY_DELTA_MB + " MB, was " + memoryDelta + " MB"); - - chunkProcessor.stop(); - } - - /** - * Tests that a processor using {@link MultiOutputIterator#setIterator(String, Iterator)} - * (independent mode) correctly streams independent lazy sources per output connection, - * equivalent to using two separate output parameters but from one - * fixed method parameter. - */ - @ParameterizedTest - @EnumSource(OutputMode.class) - void multiOutputIteratorIndependentModeShouldStreamPerConnection(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", "independentEmitter", 1, new HashMap<>()) - .orElseThrow(() -> new IllegalStateException("independentEmitter 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("MAIN", row1Struct.class); - if (outputMode == OutputMode.TWO_OUTPUT) { - outputsHandler.addConnection("REJECT", row1Struct.class); - } - } - - final InputFactory inputFactory = inputsHandler.asInputFactory(); - final OutputFactory outputFactory = outputsHandler.asOutputFactory(); - - 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); - - gc(); - final long memoryBefore = usedMemoryMB(); - - chunkProcessor.onElement(inputFactory, outputFactory); - - // Independent mode: drain each connection with hasDataFor - int mainCount = 0; - int rejectCount = 0; - if (outputMode != OutputMode.NO_OUTPUT) { - while (outputsHandler.hasMoreData()) { - if (outputsHandler.hasDataFor("MAIN")) { - final row1Struct mainRow = outputsHandler.getValue("MAIN"); - Assertions.assertNotNull(mainRow, "MAIN record should not be null"); - Assertions.assertEquals("main-" + mainCount, mainRow.id, - "MAIN record id mismatch at index " + mainCount); - mainCount++; - } - if (outputMode == OutputMode.TWO_OUTPUT && outputsHandler.hasDataFor("REJECT")) { - final row1Struct rejectRow = outputsHandler.getValue("REJECT"); - Assertions.assertNotNull(rejectRow, "REJECT record should not be null"); - Assertions.assertEquals("reject-" + rejectCount, rejectRow.id, - "REJECT record id mismatch at index " + rejectCount); - rejectCount++; - } - } - } - - chunkProcessor.flush(outputFactory); - gc(); - final long memoryDelta = usedMemoryMB() - memoryBefore; - - System.out.println("=== ProcessorBufferingTest: MultiOutputIterator independent mode ==="); - System.out.println("Output mode: " + outputMode); - System.out.println("MAIN drained: " + mainCount); - System.out.println("REJECT drained: " + rejectCount); - System.out.println("Memory delta: " + memoryDelta + " MB"); - - switch (outputMode) { - case TWO_OUTPUT: - Assertions.assertEquals(RECORD_COUNT, mainCount, - "MAIN should receive all records"); - Assertions.assertEquals(RECORD_COUNT / 2, rejectCount, - "REJECT should receive half the records"); - break; - case ONE_OUTPUT: - Assertions.assertEquals(RECORD_COUNT, mainCount, - "MAIN should receive all records"); - Assertions.assertEquals(0, rejectCount, - "REJECT not registered, should receive nothing"); - break; - case NO_OUTPUT: - Assertions.assertEquals(0, mainCount + rejectCount, - "No connections registered, no records should be received"); - break; - } - Assertions.assertTrue(memoryDelta <= MAX_MEMORY_DELTA_MB, - "Memory delta should be <= " + MAX_MEMORY_DELTA_MB + " MB, was " + memoryDelta + " MB"); - - chunkProcessor.stop(); - } - - // --- 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("FLOW") final MultiOutputIterator output) { - output.setIterator("FLOW", 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", "REJECT" }) final MultiOutputIterator out, - @LastGroup final boolean lastGroup) { - if (!lastGroup) { - return; - } - - // MAIN gets records where index % 2 != 0 - out.setIterator("MAIN", 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 - out.setIterator("REJECT", 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; - } - }); - } - } - - /** - * Processor that uses a single shared split {@link MultiOutputIterator} in {@code @AfterGroup} - * to route even-indexed records to MAIN and odd-indexed records to REJECT via {@link TaggedOutput}. - * This tests the {@code hasDataFor} fix: when only MAIN is registered, odd-indexed pending - * records must be skipped rather than stalling the drain loop. - */ - @Processor(name = "splitAfterGroupEmitter", family = "ProcessorBufferingTest") - public static class SplitAfterGroupEmitterProcessor implements Serializable { - - @BeforeGroup - public void beforeGroup() { - } - - @ElementListener - public void onElement(@Input final Record input) { - } - - @AfterGroup - public void afterGroup(@Output({ "MAIN", "REJECT" }) final MultiOutputIterator out, - @LastGroup final boolean lastGroup) { - if (!lastGroup) { - return; - } - out.setIterator(new java.util.Iterator<>() { - - private int index = 0; - - @Override - public boolean hasNext() { - return index < RECORD_COUNT; - } - - @Override - public TaggedOutput next() { - final String outputName = (index % 2 == 0) ? "MAIN" : "REJECT"; - final Record rec = builderFactory.newRecordBuilder() - .withString("id", outputName.toLowerCase() + "-" + index) - .withString("name", "split-aftergroup-record-" + index) - .withString("data", "split-aftergroup-data-" + index) - .build(); - index++; - return TaggedOutput.of(outputName, rec); - } - }); - } - } - - // --- 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()"); - } - } - - /** - * Processor that uses a single {@link MultiOutputIterator} to split 100K records - * between MAIN (even indices) and REJECT (odd indices) from a single lazy source. - * This is the true-split scenario: one source, two outputs, no buffering. - */ - @Processor(name = "splitEmitter", family = "ProcessorBufferingTest") - public static class SplitEmitterProcessor implements Serializable { - - @ElementListener - public void onElement(@Input final Record input, - @Output({ "MAIN", "REJECT" }) final MultiOutputIterator splitter) { - splitter.setIterator(new java.util.Iterator<>() { - - private int index = 0; - - @Override - public boolean hasNext() { - return index < RECORD_COUNT; - } - - @Override - public TaggedOutput next() { - final String outputName = (index % 2 == 0) ? "MAIN" : "REJECT"; - final Record rec = builderFactory.newRecordBuilder() - .withString("id", outputName.toLowerCase() + "-" + index) - .withString("name", "split-record-" + index) - .withString("data", "split-data-" + index) - .build(); - index++; - return TaggedOutput.of(outputName, rec); - } - }); - } - } - - /** - * Processor that uses {@link MultiOutputIterator#setIterator(String, Iterator)} - * (independent mode) to assign separate lazy sources to MAIN and REJECT. - * MAIN gets RECORD_COUNT records, REJECT gets RECORD_COUNT/2 records. - */ - @Processor(name = "independentEmitter", family = "ProcessorBufferingTest") - public static class IndependentEmitterProcessor implements Serializable { - - @ElementListener - public void onElement(@Input final Record input, - @Output({ "MAIN", "REJECT" }) final MultiOutputIterator out) { - out.setIterator("MAIN", new java.util.Iterator<>() { - - private int index = 0; - - @Override - public boolean hasNext() { - return index < RECORD_COUNT; - } - - @Override - public Record next() { - return builderFactory.newRecordBuilder() - .withString("id", "main-" + index) - .withString("name", "main-record-" + index++) - .build(); - } - }); - out.setIterator("REJECT", new java.util.Iterator<>() { - - private int index = 0; - - @Override - public boolean hasNext() { - return index < RECORD_COUNT / 2; - } - - @Override - public Record next() { - return builderFactory.newRecordBuilder() - .withString("id", "reject-" + index) - .withString("name", "reject-record-" + index++) - .build(); - } - }); - } - } - - 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 6aa430433b7c7..d4aa89522267a 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 or MultiOutputIterator + - @Output parameter must be of type OutputEmitter - 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"""); }