Yyin/QTDI-653 add LastGroup annotation to Processor's afterGroup method.

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

@@ -0,0 +1,27 @@
+/**
+ * Copyright (C) 2006-2024 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 static java.lang.annotation.ElementType.PARAMETER;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+@Target(PARAMETER)
+@Retention(RUNTIME)
+public @interface LastGroup {
+}

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

@@ -26,5 +26,11 @@ public interface Processor extends Lifecycle {
     // since it will never work in the studio with current generation logic
     void afterGroup(OutputFactory output);
 
+    default void afterGroup(OutputFactory output, boolean last) {
+        afterGroup(output);
+    }
+
+    default boolean isLastGroupUsed() { return false; }
+
     void onNext(InputFactory input, OutputFactory output);
 }

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

@@ -33,6 +33,7 @@
 import java.lang.reflect.ParameterizedType;
 import java.util.AbstractMap;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -53,6 +54,7 @@
 import org.talend.sdk.component.api.processor.BeforeGroup;
 import org.talend.sdk.component.api.processor.ElementListener;
 import org.talend.sdk.component.api.processor.Input;
+import org.talend.sdk.component.api.processor.LastGroup;
 import org.talend.sdk.component.api.processor.Output;
 import org.talend.sdk.component.api.service.record.RecordBuilderFactory;
 import org.talend.sdk.component.runtime.base.Delegated;
@@ -122,7 +124,8 @@ public void beforeGroup() {
                     : Stream.of(process.getParameters()).map(this::buildProcessParamBuilder).collect(toList());
             parameterBuilderAfterGroup = afterGroup
                     .stream()
-                    .map(after -> new AbstractMap.SimpleEntry<>(after, Stream.of(after.getParameters()).map(param -> {
+                    .map(after -> new AbstractMap.SimpleEntry<>(after, Stream.of(after.getParameters())
+                            .map(param -> {
                         if (isGroupBuffer(param.getParameterizedType())) {
                             expectedRecordType = Class.class
                                     .cast(ParameterizedType.class
@@ -162,6 +165,9 @@ private BiFunction<InputFactory, OutputFactory, Object> buildProcessParamBuilder
 
     private Function<OutputFactory, Object> toOutputParamBuilder(final Parameter parameter) {
         return outputs -> {
+            if (parameter.isAnnotationPresent(LastGroup.class)) {
+                return false;
+            }
             final String name = parameter.getAnnotation(Output.class).value();
             return outputs.create(name);
         };
@@ -239,13 +245,40 @@ private JsonProvider jsonProvider() {
 
     @Override
     public void afterGroup(final OutputFactory output) {
-        afterGroup
-                .forEach(after -> doInvoke(after,
-                        parameterBuilderAfterGroup
-                                .get(after)
-                                .stream()
-                                .map(b -> b.apply(output))
-                                .toArray(Object[]::new)));
+        afterGroup.forEach(after -> {
+            Object[] params = parameterBuilderAfterGroup.get(after)
+                    .stream()
+                    .map(b -> b.apply(output))
+                    .toArray(Object[]::new);
+            doInvoke(after, params);
+        });
+        if (records != null) {
+            records = null;
+        }
+    }
+
+    @Override
+    public boolean isLastGroupUsed() {
+        for (Method after : afterGroup) {
+            for (Parameter param : after.getParameters()) {
+                if (param.isAnnotationPresent(LastGroup.class)) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
+    @Override
+    public void afterGroup(final OutputFactory output, final boolean last) {
+        afterGroup.forEach(after -> {
+            Object[] params = Stream.concat(
+                    parameterBuilderAfterGroup.get(after).stream()
+                            .map(b -> b.apply(output))
+                            .filter(b -> !b.equals(false)),
+                    Stream.of(last)).toArray(Object[]::new);
+            doInvoke(after, params);
+        });
         if (records != null) {
             records = null;
         }

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

@@ -45,6 +45,7 @@
 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.LastGroup;
 import org.talend.sdk.component.api.processor.Output;
 import org.talend.sdk.component.api.processor.OutputEmitter;
 import org.talend.sdk.component.api.processor.Processor;
@@ -199,13 +200,34 @@ private void validateProcessor(final Class<?> input) {
                 }
             })
                     .filter(p -> !p.isAnnotationPresent(Output.class))
+                    .filter(p -> !p.isAnnotationPresent(LastGroup.class))
                     .filter(p -> !Parameters.isGroupBuffer(p.getParameterizedType()))
                     .collect(toList());
             if (!invalidParams.isEmpty()) {
                 throw new IllegalArgumentException("Parameter of AfterGroup method need to be annotated with Output");
             }
         });
+        if (afterGroups
+                .stream()
+                .anyMatch(m -> Stream.of(m.getParameters()).anyMatch(p -> p.isAnnotationPresent(LastGroup.class)))
+                && afterGroups.size() > 1) {
+            throw new IllegalArgumentException(input
+                    + " must have a single @AfterGroup method with @LastGroup parameter");
+        }
+
+        validateProducer(input, afterGroups);
+
+        Stream.of(input.getMethods()).filter(m -> m.isAnnotationPresent(BeforeGroup.class)).forEach(m -> {
+            if (m.getParameterCount() > 0) {
+                throw new IllegalArgumentException(m + " must not have any parameter");
+            }
+        });
 
+        validateAfterVariableAnnotationDeclaration(input);
+        validateAfterVariableContainer(input);
+    }
+
+    private void validateProducer(final Class<?> input, final List<Method> afterGroups) {
         final List<Method> producers = Stream
                 .of(input.getMethods())
                 .filter(m -> m.isAnnotationPresent(ElementListener.class))
@@ -227,15 +249,6 @@ private void validateProcessor(final Class<?> input) {
         }).filter(p -> !p.isAnnotationPresent(Output.class)).count() < 1) {
             throw new IllegalArgumentException(input + " doesn't have the input parameter on its producer method");
         }
-
-        Stream.of(input.getMethods()).filter(m -> m.isAnnotationPresent(BeforeGroup.class)).forEach(m -> {
-            if (m.getParameterCount() > 0) {
-                throw new IllegalArgumentException(m + " must not have any parameter");
-            }
-        });
-
-        validateAfterVariableAnnotationDeclaration(input);
-        validateAfterVariableContainer(input);
     }
 
     private boolean validOutputParam(final Parameter p) {

component-runtime-impl/src/test/java/org/talend/sdk/component/runtime/output/ProcessorImplTest.java

@@ -21,8 +21,10 @@
 import static java.util.Collections.singletonList;
 import static java.util.stream.Collectors.toList;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotSame;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.IOException;
 import java.io.Serializable;
@@ -37,6 +39,9 @@
 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.LastGroup;
+import org.talend.sdk.component.api.processor.Output;
+import org.talend.sdk.component.api.processor.OutputEmitter;
 import org.talend.sdk.component.api.record.Record;
 import org.talend.sdk.component.runtime.record.RecordImpl;
 import org.talend.sdk.component.runtime.serialization.Serializer;
@@ -64,12 +69,24 @@ void bulkGroup() {
             data.forEach(it -> processor.onNext(n -> it, null));
             assertNull(Bufferized.RECORDS);
             processor.afterGroup(null);
+            assertFalse(processor.isLastGroupUsed());
             assertEquals(data, Bufferized.RECORDS);
             Bufferized.RECORDS = null;
         }
         processor.stop();
     }
 
+    @Test
+    void bulkGroupWithLastGroup() {
+        final Processor processor = new ProcessorImpl("Root", "Test", "Plugin", emptyMap(), new SampleLastGroupOutput());
+        processor.start();
+        processor.beforeGroup();
+        assertTrue(processor.isLastGroupUsed());
+        processor.afterGroup(NO_OUTPUT, true);
+        assertTrue(SampleLastGroupOutput.isCalled);
+        processor.stop();
+    }
+
     @Test
     void lifecycle() {
         assertLifecycle(new SampleProcessor());
@@ -186,4 +203,19 @@ public static class Sample {
 
         private int data;
     }
+
+    public static class SampleLastGroupOutput  implements Serializable {
+        private static boolean isCalled = false;
+
+        @ElementListener
+        public void onNext(final Sample sample) {
+
+        }
+
+        @AfterGroup
+        public void afterGroup(@Output("REJECT") final OutputEmitter<Record> records, @LastGroup final boolean isLast) {
+            isCalled = isLast;
+        }
+    }
+
 }

component-runtime-impl/src/test/java/org/talend/sdk/component/runtime/visitor/visitor/ModelVisitorTest.java

@@ -40,6 +40,9 @@
 import org.talend.sdk.component.api.input.Split;
 import org.talend.sdk.component.api.processor.AfterGroup;
 import org.talend.sdk.component.api.processor.ElementListener;
+import org.talend.sdk.component.api.processor.LastGroup;
+import org.talend.sdk.component.api.processor.Output;
+import org.talend.sdk.component.api.processor.OutputEmitter;
 import org.talend.sdk.component.api.processor.Processor;
 import org.talend.sdk.component.api.record.Record;
 import org.talend.sdk.component.api.standalone.DriverRunner;
@@ -65,6 +68,18 @@ void validBulk() {
         assertEquals(singletonList(
                 "@Processor(org.talend.sdk.component.runtime.visitor.visitor.ModelVisitorTest$ProcessorBulk$Out)"),
                 visit(ProcessorBulk.class));
+
+    }
+    @Test
+    void validAfterGroupWithLastGroup() {
+        assertEquals(singletonList(
+                        "@Processor(org.talend.sdk.component.runtime.visitor.visitor.ModelVisitorTest$ProcessorOneAfterGroup$Out)"),
+                visit(ProcessorOneAfterGroup.class));
+    }
+
+    @Test
+    void afterGroupWithLastGroupMoreThanOne() {
+        assertThrows(IllegalArgumentException.class, () -> visit(ProcessorTwoAfterGroup.class));
     }
 
     @Test
@@ -589,6 +604,39 @@ public void commit(final Collection<Record> records) {
         }
     }
 
+    public static class ProcessorOneAfterGroup {
+
+        @Processor(family = "comp", name = "Bulk")
+        public static class Out {
+
+            @ElementListener
+            public void onNext(final In in) {
+                // no-op
+            }
+            @AfterGroup
+            public void afterGroup(@Output("REJECT") final OutputEmitter<Record> rejected, @LastGroup final boolean isLast) {
+                // no-op
+            }
+        }
+    }
+
+    public static class ProcessorTwoAfterGroup {
+
+        @Processor(family = "comp", name = "Bulk")
+        public static class Out {
+
+            @AfterGroup
+            public void commit(final Collection<Record> records) {
+                // no-op
+            }
+
+            @AfterGroup
+            public void afterGroup(@Output("REJECT") final OutputEmitter<Record> rejected, @LastGroup final boolean isLast) {
+                // no-op
+            }
+        }
+    }
+
     public static class EmitterNoProduces {
 
         @Emitter(family = "comp", name = "Input")

component-runtime-manager/src/main/java/org/talend/sdk/component/runtime/manager/chain/AutoChunkProcessor.java

@@ -25,11 +25,11 @@
 @RequiredArgsConstructor
 public class AutoChunkProcessor implements Lifecycle {
 
-    private final int chunkSize;
+    protected final int chunkSize;
 
-    private final Processor processor;
+    protected final Processor processor;
 
-    private int processedItemCount = 0;
+    protected int processedItemCount = 0;
 
     public void onElement(final InputFactory ins, final OutputFactory outs) {
         if (processedItemCount == 0) {

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

@@ -15,6 +15,8 @@
  */
 package org.talend.sdk.component.runtime.di;
 
+import org.talend.sdk.component.runtime.output.InputFactory;
+import org.talend.sdk.component.runtime.output.OutputFactory;
 import org.talend.sdk.component.runtime.output.Processor;
 
 /*
@@ -26,4 +28,31 @@ public class AutoChunkProcessor extends org.talend.sdk.component.runtime.manager
     public AutoChunkProcessor(final int chunkSize, final Processor processor) {
         super(chunkSize, processor);
     }
+
+    @Override
+    public void onElement(final InputFactory ins, final OutputFactory outs) {
+        if (processedItemCount == 0) {
+            processor.beforeGroup();
+        }
+        try {
+            processor.onNext(ins, outs);
+            processedItemCount++;
+        } finally {
+            if (processedItemCount == chunkSize) {
+                processor.afterGroup(outs);
+                processedItemCount = 0;
+            }
+        }
+    }
+
+    @Override
+    public void flush(final OutputFactory outs) {
+        if (processor.isLastGroupUsed()) {
+            processor.afterGroup(outs, true);
+        }
+        if (processedItemCount > 0) {
+            processor.afterGroup(outs);
+        }
+        processedItemCount = 0;
+    }
 }