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 ?

… 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.

thboileau · 2026-07-13T13:03:19Z


        private final Class<T> type;

+        private Iterator<T> source;


I feel we need more documentation here.

First, the IO class is shared between the InputsHandler and the OutputsHandler classes, and implements a "push-or-pull" behavior which is only used by the OutputsHandler. People may read InputsHandler and wonder why "source" is never used. We should document it.

Secondly, as we know that "push-or-pull" is actually exclusive, we could think of a better design that clearly separates these two behaviors, but I don't how, I'm not smart enough. When IO is created, there is no notion of "push" or "pull", just the name of the connection. So documentation would help here

What do you think about this?

about refactor, no code compatiable risk for future as that is not so public to client user, we can do later. So agree document enough now.

… OutputsHandler

… in component structure

wwang-talend · 2026-07-14T07:13:43Z

            return type;
        }
+
+        private void closeSource() {


the autoclose not work yet, except we call that .reset() method some where in studio common javajet, but in fact, not. So i may remove this part.

for httpclient==>
every input record==> make M(bigger enough to use streaming feature here) output records

we may need to use this autoclose to release the resource for pre streaming object asap? not wait on @PreDestory close method.

But studio allow user not draw output line even allow. So for safe, better to close in @PreDestory too. Above one for "close asap", @PreDestory for "close finally even outside exception happen and so on"

so keep it, not public and not make code refactor complex in future.

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.

sonar-rnd · 2026-07-16T03:00:42Z

Quality Gate failed

0.00% Coverage on New Code (is less than 80.00%)
2 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

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

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

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!

thboileau Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

wwang-talend Jul 13, 2026

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!

wwang-talend Jul 14, 2026

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!

wwang-talend Jul 14, 2026

Choose a reason for hiding this comment

Uh oh!

wwang-talend Jul 14, 2026

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!

wwang-talend Jul 14, 2026

Choose a reason for hiding this comment

Uh oh!

wwang-talend commented Jul 14, 2026

Uh oh!

sonar-rnd Bot commented Jul 16, 2026

Quality Gate failed

Uh oh!

Reviewers

Assignees

Labels

🔴 Critical: `ClassCastException` in non-DI contexts

Clarification on `DataOutputFactory` / `JobImpl`

wwang-talend commented Jul 13, 2026 •

edited

Loading

thboileau Jul 13, 2026 •

edited

Loading

wwang-talend Jul 14, 2026 •

edited

Loading

thboileau Jul 13, 2026 •

edited

Loading

wwang-talend Jul 14, 2026 •

edited

Loading

wwang-talend Jul 14, 2026 •

edited

Loading