fix(QTDI-2709): TCK processors should send generated records to next connector without buffer by wwang-talend · Pull Request #1253 · Talend/component-runtime

wwang-talend · 2026-07-13T06:49:54Z

Requirements

Any code change adding any logic MUST be tested through a unit test executed with the default build
Any API addition MUST be done with a documentation update if relevant

this PR has been written with the help of GitHub Copilot or another generative AI tool

…tion during record emission

wwang-talend

Code Review — PR #1253

🔴 Critical: `ClassCastException` in non-DI contexts

ProcessorImpl.buildProcessParamBuilder() and toOutputParamBuilder() cast outputs.create(name) to OutputIterator:

// ProcessorImpl.java:156
return (inputs, outputs) -> (OutputIterator) outputs.create(name);

But OutputFactory.create() returns OutputEmitter. Only the DI/Studio OutputsHandler returns objects implementing both interfaces. Two other OutputFactory implementations do NOT implement OutputIterator:

BeamOutputFactory / BeamOutputEmitter (BaseProcessorFn.java:162-176)
DataOutputFactory / OutputEmitterImpl (JobImpl.java:661-681)

Any processor using @Output OutputIterator<Record> in a Beam runner or via JobImpl chain will throw ClassCastException at runtime.

Suggested fix: Either update OutputFactory to provide OutputIterator support (e.g., a default method), or have ProcessorImpl wrap the OutputEmitter in an adapter, or update all OutputFactory implementations.

🟡 Design: Implicit contract not enforced by type system

The OutputFactory interface declares OutputEmitter create(String name). This PR adds an implicit requirement that returned objects also implement OutputIterator, but nothing enforces this. Future OutputFactory implementations will hit the same ClassCastException silently at runtime.

🟡 Raw types throughout

OutputEmitterWithIterator implements raw OutputEmitter and OutputIterator (no type parameters). setIterator() takes raw Iterator. This bypasses generic type safety and produces compiler warnings.

🟡 Silent exception swallowing

BaseIOHandler.IO.closeSource() catches all exceptions with an empty block:

} catch (final Exception e) {
    // best effort cleanup
}

Consider at minimum logging at DEBUG/TRACE level to avoid hiding bugs during iterator cleanup.

🟡 Test reliability concerns

Memory-based assertions using System.gc() + Runtime.freeMemory() are non-deterministic. The 5MB threshold could be flaky across JVMs, GC algorithms, or CI environments.
builderFactory is a protected static mutable field — risky with parallel test execution.

wwang-talend · 2026-07-13T07:15:07Z

OutputIterator only for studio, only studio have that case, only need to document it that not suitable for beam env.

wwang-talend

Response to AI Review

🔴 ClassCastException in non-DI contexts

Valid observation. However, this is by design — the OutputIterator pattern was built specifically for the Studio DI runtime to solve the single-threaded buffering problem.

As stated in the spec: "The impact of this pattern on Cloud is minimal. In the Cloud, processors are not TCK processors (usually dataprep functions)."

No real-world processor will use @Output OutputIterator<T> in a Beam pipeline. Adding adapter/fallback code in BeamOutputFactory / JobImpl would add unnecessary complexity for a use case that does not exist.

Action taken: Added explicit javadoc on OutputIterator documenting that it is Studio DI only and not supported in Beam runners.

🟡 Implicit contract not enforced by type system

Accepted trade-off. We control all OutputFactory implementations. The ModelVisitor validates processor declarations at plugin load time, so misuse surfaces early. A default method on OutputFactory would be cleaner but is a bigger change with no practical benefit given the Studio-only scope.

🟡 Raw types

Fair point. Will address in a follow-up cleanup.

🟡 Silent exception swallowing

Agreed. Will add logging in the catch block.

🟡 Test reliability

The memory test is intentionally designed as a manual/observable test — it prints memory stats so developers can see the delta. The 5MB threshold is generous (actual delta is ~1-2MB with iterator mode). The static builderFactory pattern is consistent with existing tests (DIBulkAutoChunkTest, DIBatchSimulationTest).

wwang-talend · 2026-07-13T07:19:32Z

Clarification on `DataOutputFactory` / `JobImpl`

The reviewer flagged DataOutputFactory (JobImpl.java:661-681) as another non-DI OutputFactory that would throw ClassCastException.

Analysis: JobImpl with DataOutputFactory is the local chain execution path used by the Job.builder() programmatic API (component-runtime-manager). It is NOT Studio DI and NOT Beam.

However, this API is only referenced in documentation examples — no production usage exists. The real Studio DI runtime uses OutputsHandler which correctly implements both OutputEmitter and OutputIterator.

The OutputIterator javadoc already clarifies it is supported in Studio DI runtime only. This is sufficient — adding adapter code in JobImpl.DataOutputFactory would be unnecessary complexity for a path that will not encounter OutputIterator processors in practice.

…e support limitations

wwang-talend · 2026-07-13T08:47:13Z

sonar often report some small smell in tests, not worth to work on it even, should care on the major things we need to care thing, not that smell even in junit tests only.

Every thing good smell, but bug appear. That is a kind of context go out like AI.

thboileau · 2026-07-13T09:10:39Z

+            return null;
+        } else if (value instanceof javax.json.JsonValue) {
+            return jsonb.fromJson(value.toString(), ref.getType());
+        } else if (value instanceof Record record) {


use rec for instance ?

i revert it as "Record record" is right expression, here i not agree sonar.

… and ProcessorBufferingTest

thboileau · 2026-07-13T12:38:35Z

-            };
+            final String name = parameter.getAnnotation(Output.class).value();
+            if (OutputIterator.class == parameter.getType()) {
+                return (inputs, outputs) -> (OutputIterator) outputs.create(name);


That only works because the one concrete implementation, OutputsHandler.OutputEmitterWithIterator, happens to implement both interfaces. Nothing in OutputFactory's signature says a factory may hand back something iterator-capable.

One solution is to make the capability explicit in the OutputFactory : a default OutputIterator createIterator(String name) that throws a clear UnsupportedOperationException by default, except in the case of the OutputsHandler class.

The default implementation of createIterator allows retro-compatibility

public interface OutputFactory { OutputEmitter create(String name); default OutputIterator createIterator(String name) { throw new UnsupportedOperationException(); } }

In OutputsHandler:

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 OutputIterator createIterator(final String name) { final BaseIOHandler.IO ref = connections.get(getActualName(name)); return iterator -> { if (ref == null) { return; } ref.setSource(new Iterator() { @Override public boolean hasNext() { return iterator.hasNext(); } @Override public Object next() { return convert(iterator.next(), ref); } }); }; } }; }

What do you think about this proposal?

Hi @thboileau , thanks for the review.

OutputFactory is a public interface, but not so public as only studio common javajet use it. So from the history regression risk for future view, i think your idea is ok, which add a new interface method.

My origin idea is not change interface, keep complex in implement.

So i will consider your advice as it look good, but give me some times to prove your idea is real better.

… OutputsHandler

… in component structure

wwang-talend · 2026-07-14T07:28:06Z

about parallize, i not find issue for currernt design for studio part. But need to test

This reverts commit 25264bc.

…lazy sources

…r in documentation and tests

…and improve split streaming tests

…ecord validity

…related processing logic

…multiple output values

iteration logic

thboileau · 2026-07-21T13:11:34Z

+ * The drain loop on the Studio side uses {@code hasDataFor(name)} for correct
+ * per-connection checking in both modes:
+ *
+ * <pre>


I think that this part does not relate directly to the MultiIterator.
It's part of the API of BaseIOHandler class, can be implemented based on the current state of BaseIOHandler (ie.: with only the support of OutputEmitter). It's a nice replacement of the if (row2 != null) { that we can find in the generated code.

while (outputHandler_tDateFormatValidatorValidator_1.hasMoreData()) { nbLineOutput_tDateFormatValidatorValidator_1++; row2 = outputHandler_tDateFormatValidatorValidator_1.getValue("FLOW"); row3 = outputHandler_tDateFormatValidatorValidator_1.getValue("reject"); /** * [tDateFormatValidatorValidator_1 process_data_begin ] stop */ // Start of branch "row2" if (row2 != null) { /** * [tLogRow_1 main ] start */

that is used here : https://github.com/Talend/studio/pull/1672/changes#diff-803bb6ddc517da716b80b39a1dffd50b4fc5b3fe14b4c65c3dbca953bcc754f6L35

about this code generation, the patterns :

pattern 1: one input record come ==> dispatch output record, go output 1(main) or go output 2(reject) or output xxx, but only go unique output x.

pattern 2: any, group or not group, or one by one ==> output x records to output 1(main), output y to output 2(reject), z to output 3...,x, y, z value even no relationship with input records number

About pattern 1, the code generation above is right.

About pattern 2, the code generation above is strange.

TODO desc more

What I want to say is that documenting outputsHandler.HasMoreData is right in class BaseIOHandler, not MultiOutputIterator.

aha, i will correct the doc, thanks. TODO

thboileau · 2026-07-21T16:29:52Z

+     * @param outputName the output connection name (e.g. {@code "MAIN"}, {@code "REJECT"})
+     * @param iterator the lazy iterator producing records for that connection
+     */
+    void setIterator(String outputName, Iterator<T> iterator);


I really wonder if we should drop this in another interface. It's another user case.

that for the case which no need use TaggedOutput, like httpclient. And also sure httpclient can use TaggedOutput.

yes, that's why I prefer another interface, like SingleOutputIterator for instance

image one day:
no input==>httpclient ==> main
==> reject
==> xxxx

a number input==>httpclient ==> x number main
==> y reject
==> z xxxx
a value no relationship with x + y + z and so on.

MultiOutputIterator support one channel too. But SingleOutputIterator don't support multi channel.

yes, I understand. to me these are two different uses case that could be handled by tho distinct interfaces.

nb: let's start with your solution. I propose to discuss the topic with Yves and Manu when they will be back.

thboileau · 2026-07-22T06:35:35Z

+                            }
+                            final TaggedOutput<T> tagged = iterator.next();
+                            final String name = tagged.getOutputName();
+                            final BaseIOHandler.IO ref = connections.get(name);


Suggested change

final BaseIOHandler.IO ref = connections.get(name);

final BaseIOHandler.IO ref = connections.get(getActualName(name));

i remove getActualName on purpose, as no meaning, and a equals method in a method called often.

that just convert "default" to "FLOW", connector developer never use "default", that is a hiden thing in connector runtime.

but maybe over consider performance

in this case, shouldn't we drop the other references to getActualName ? Either we keep getActualName or we drop it. we can't be in the middle because it makes maintenance difficult.

… enhanced output handling

sonar-rnd · 2026-07-22T08:11:23Z

Quality Gate failed

0.00% Coverage on New Code (is less than 80.00%)
12 New Issues (is greater than 0)

Project ID: org.talend.sdk.component:component-runtime

View in SonarQube

wwang-talend added 3 commits July 13, 2026 11:43

fix(QTDI-2709): Add ProcessorBufferingTest to evaluate memory consump…

9853869

…tion during record emission

fix(QTDI-2709): do real implement

9aecf24

fix(QTDI-2709): remove verbose iterator attribute

b33e762

wwang-talend commented Jul 13, 2026

View reviewed changes

doc(QTDI-2709): update OutputIterator documentation to clarify runtim…

87deb2d

…e support limitations

wwang-talend requested a review from thboileau July 13, 2026 08:33

thboileau reviewed Jul 13, 2026

View reviewed changes

wwang-talend added 2 commits July 13, 2026 17:20

fix(QTDI-2709): refactor variable names for clarity in OutputsHandler…

9e8138b

… and ProcessorBufferingTest

Merge branch 'master' into wwang-talend/QTDI-2709-streaming

1317b18

wwang-talend requested a review from thboileau July 13, 2026 11:05

thboileau reviewed Jul 13, 2026

View reviewed changes

Comment thread ...io/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/BaseIOHandler.java

fix(QTDI-2709): implement OutputIterator support in OutputFactory and…

0c7f73c

… OutputsHandler

wwang-talend requested a review from thboileau July 14, 2026 03:22

fix(QTDI-2709): rename ProcessorBufferingTest package for consistency…

cac943b

… in component structure

thboileau previously approved these changes Jul 14, 2026

View reviewed changes

wwang-talend commented Jul 14, 2026

View reviewed changes

Comment thread ...io/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/BaseIOHandler.java

wwang-talend added 4 commits July 16, 2026 10:11

add streaming api

25264bc

improve test

1788396

Revert "add streaming api"

283bd80

This reverts commit 25264bc.

improve test

d7273b4

wwang-talend dismissed thboileau’s stale review via d7273b4 July 16, 2026 02:34

wwang-talend added 3 commits July 16, 2026 17:27

feat: implement MultiOutputIterator for split streaming support

d2124c6

refactor: update tagged source handling for split streaming support

386bb57

feat: enhance MultiOutputIterator with independent mode for separate …

d54c03f

…lazy sources

wwang-talend added 7 commits July 16, 2026 18:03

feat: enhance MultiOutputIterator with independent mode for separate …

d9391c2

…lazy sources

refactor: update references from OutputIterator to MultiOutputIterato…

b5bfa04

…r in documentation and tests

feat: enhance hasDataFor method to skip unregistered pending records …

cbc6b10

…and improve split streaming tests

test: enhance ProcessorBufferingTest with additional assertions for r…

f0b35b3

…ecord validity

feat: update Output annotation to support multiple values and adjust …

df17146

…related processing logic

feat: update Output annotations in ProcessorBufferingTest to support …

6cc661d

…multiple output values

refactor: simplify IO handling by removing caching and optimizing

94ee246

iteration logic

wwang-talend requested a review from thboileau July 20, 2026 10:27

thboileau reviewed Jul 21, 2026

View reviewed changes

Comment thread ...-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/schema/TaCoKitGuessSchema.java

thboileau reviewed Jul 21, 2026

View reviewed changes

Comment thread ...io/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/BaseIOHandler.java

thboileau reviewed Jul 21, 2026

View reviewed changes

Comment thread ...o/component-runtime-di/src/main/java/org/talend/sdk/component/runtime/di/OutputsHandler.java

thboileau reviewed Jul 21, 2026

View reviewed changes

thboileau reviewed Jul 22, 2026

View reviewed changes

feat: integrate MultiOutputIterator support in TaCoKitGuessSchema for…

07d3064

… enhanced output handling

	final BaseIOHandler.IO ref = connections.get(name);
	final BaseIOHandler.IO ref = connections.get(getActualName(name));

Uh oh!

Conversation

wwang-talend commented Jul 13, 2026

Requirements

Why this PR is needed?

What does this PR adds (design/code thoughts)?

AI generated code

Uh oh!

wwang-talend left a comment

Choose a reason for hiding this comment

Code Review — PR #1253

🔴 Critical: ClassCastException in non-DI contexts

🟡 Design: Implicit contract not enforced by type system

🟡 Raw types throughout

🟡 Silent exception swallowing

🟡 Test reliability concerns

Uh oh!

wwang-talend commented Jul 13, 2026

Uh oh!

wwang-talend left a comment

Choose a reason for hiding this comment

Response to AI Review

🔴 ClassCastException in non-DI contexts

🟡 Implicit contract not enforced by type system

🟡 Raw types

🟡 Silent exception swallowing

🟡 Test reliability

Uh oh!

wwang-talend commented Jul 13, 2026

Clarification on DataOutputFactory / JobImpl

Uh oh!

wwang-talend commented Jul 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wwang-talend Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

thboileau Jul 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wwang-talend Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

wwang-talend commented Jul 14, 2026

Uh oh!

Uh oh!

Uh oh!

thboileau Jul 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wwang-talend Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

thboileau Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wwang-talend Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

🔴 Critical: `ClassCastException` in non-DI contexts

Clarification on `DataOutputFactory` / `JobImpl`

wwang-talend commented Jul 13, 2026 •

edited

Loading

wwang-talend Jul 22, 2026 •

edited

Loading

thboileau Jul 13, 2026 •

edited

Loading

wwang-talend Jul 14, 2026 •

edited

Loading

thboileau Jul 21, 2026 •

edited

Loading

wwang-talend Jul 22, 2026 •

edited

Loading

thboileau Jul 22, 2026 •

edited

Loading

wwang-talend Jul 22, 2026 •

edited

Loading

wwang-talend Jul 22, 2026 •

edited

Loading

wwang-talend Jul 22, 2026 •

edited

Loading

thboileau Jul 22, 2026 •

edited

Loading

wwang-talend Jul 22, 2026 •

edited

Loading

thboileau Jul 22, 2026 •

edited

Loading