NIFI-12757: Issue GC commands to Python for FlowFileTransformResults and RecordTransformResults when no longer needed on Java side

This closes #8375

Signed-off-by: David Handermann <exceptionfactory@apache.org>

nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/client/NiFiPythonGateway.java

@@ -27,6 +27,7 @@ import java.util.Stack;
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.nifi.python.PythonController;
+import org.apache.nifi.python.PythonObjectProxy;
 import org.apache.nifi.python.processor.PreserveJavaBinding;
 import org.apache.nifi.python.processor.PythonProcessor;
 import org.slf4j.Logger;
@@ -142,9 +143,23 @@ public class NiFiPythonGateway extends Gateway {
     protected PythonProxyHandler createPythonProxyHandler(final String id) {
         logger.debug("Creating Python Proxy Handler for ID {}", id);
         final PythonProxyInvocationHandler createdHandler = new PythonProxyInvocationHandler(this, id);
+
+        Method proxyFreeMethod;
+        try {
+            proxyFreeMethod = PythonObjectProxy.class.getMethod("free");
+        } catch (final NoSuchMethodException ignored) {
+            proxyFreeMethod = null;
+        }
+
+        final Method freeMethod = proxyFreeMethod;
         return new PythonProxyHandler(id, this) {
             @Override
             public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
+                if (Objects.equals(freeMethod, method)) {
+                    createdHandler.free();
+                    return null;
+                }
+
                 return createdHandler.invoke(proxy, method, args);
             }
 

nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/client/PythonProxyInvocationHandler.java

@@ -37,11 +37,20 @@ public class PythonProxyInvocationHandler implements InvocationHandler {
     private final String objectId;
     private final NiFiPythonGateway gateway;
     private final JavaObjectBindings bindings;
+    private final String gcCommand;
 
     public PythonProxyInvocationHandler(final NiFiPythonGateway gateway, final String objectId) {
         this.objectId = objectId;
         this.gateway = gateway;
         this.bindings = gateway.getObjectBindings();
+        this.gcCommand = "g\n" + objectId + "\ne\n";
+    }
+
+    public void free() {
+        if (this.objectId != Protocol.ENTRY_POINT_OBJECT_ID) {
+            logger.debug("Issuing GC command to python for proxy id {}", this.objectId);
+            gateway.getCallbackClient().sendCommand(gcCommand, false);
+        }
     }
 
     @Override

nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/FlowFileTransformProxy.java

@@ -72,26 +72,31 @@ public class FlowFileTransformProxy extends PythonProcessorProxy {
             session.transfer(original, REL_FAILURE);
             return;
         }
-        final String relationshipName = result.getRelationship();
-        final Relationship relationship = new Relationship.Builder().name(relationshipName).build();
-        if (REL_FAILURE.getName().equals(relationshipName)) {
-            session.remove(transformed);
-            session.transfer(original, REL_FAILURE);
-            return;
-        }
 
-        final Map<String, String> attributes = result.getAttributes();
+        try {
-        if (attributes != null) {
+            final String relationshipName = result.getRelationship();
-            transformed = session.putAllAttributes(transformed, attributes);
+            final Relationship relationship = new Relationship.Builder().name(relationshipName).build();
-        }
+            if (REL_FAILURE.getName().equals(relationshipName)) {
+                session.remove(transformed);
+                session.transfer(original, REL_FAILURE);
+                return;
+            }
 
-        final byte[] contents = result.getContents();
+            final Map<String, String> attributes = result.getAttributes();
-        if (contents != null) {
+            if (attributes != null) {
-            transformed = session.write(transformed, out -> out.write(contents));
+                transformed = session.putAllAttributes(transformed, attributes);
-        }
+            }
 
-        session.transfer(transformed, relationship);
+            final byte[] contents = result.getContents();
-        session.transfer(original, REL_ORIGINAL);
+            if (contents != null) {
+                transformed = session.write(transformed, out -> out.write(contents));
+            }
+
+            session.transfer(transformed, relationship);
+            session.transfer(original, REL_ORIGINAL);
+        } finally {
+            result.free();
+        }
     }
 
 }

nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/FlowFileTransformResult.java

@@ -17,9 +17,11 @@
 
 package org.apache.nifi.python.processor;
 
+import org.apache.nifi.python.PythonObjectProxy;
+
 import java.util.Map;
 
-public interface FlowFileTransformResult {
+public interface FlowFileTransformResult extends PythonObjectProxy {
     String getRelationship();
 
     byte[] getContents();

nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/RecordTransformProxy.java

@@ -147,9 +147,13 @@ public class RecordTransformProxy extends PythonProcessorProxy {
                         baos.reset();
 
                         final List<RecordTransformResult> results = transform.transformRecord(json, recordSchema, attributeMap);
-                        for (final RecordTransformResult result : results) {
+                        try {
-                            writeResult(result, destinationTuples, writerFactory, session, flowFile);
+                            for (final RecordTransformResult result : results) {
-                            recordsWritten++;
+                                writeResult(result, destinationTuples, writerFactory, session, flowFile);
+                                recordsWritten++;
+                            }
+                        } finally {
+                            results.forEach(RecordTransformResult::free);
                         }
 
                         writtenSinceFlush = 0;
@@ -163,9 +167,13 @@ public class RecordTransformProxy extends PythonProcessorProxy {
                     baos.reset();
 
                     final List<RecordTransformResult> results = transform.transformRecord(json, recordSchema, attributeMap);
-                    for (final RecordTransformResult result : results) {
+                    try {
-                        writeResult(result, destinationTuples, writerFactory, session, flowFile);
+                        for (final RecordTransformResult result : results) {
-                        recordsWritten++;
+                            writeResult(result, destinationTuples, writerFactory, session, flowFile);
+                            recordsWritten++;
+                        }
+                    } finally {
+                        results.forEach(RecordTransformResult::free);
                     }
                 }
             }

nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/RecordTransformResult.java

@@ -17,11 +17,12 @@
 
 package org.apache.nifi.python.processor;
 
+import org.apache.nifi.python.PythonObjectProxy;
 import org.apache.nifi.serialization.record.RecordSchema;
 
 import java.util.Map;
 
-public interface RecordTransformResult {
+public interface RecordTransformResult extends PythonObjectProxy {
 
     String getRecordJson();
 

nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonObjectProxy.java

@@ -15,14 +15,8 @@
  * limitations under the License.
  */
 
-package org.apache.nifi.python.processor;
+package org.apache.nifi.python;
 
-import org.apache.nifi.serialization.record.RecordSchema;
+public interface PythonObjectProxy {
-
+    void free();
-public interface OutputRecord {
-    String getRelationship();
-
-    Object getRecord();
-
-    RecordSchema getSchema();
 }

nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonProcessorDetails.java

@@ -23,7 +23,7 @@ import org.apache.nifi.python.processor.documentation.UseCaseDetails;
 
 import java.util.List;
 
-public interface PythonProcessorDetails {
+public interface PythonProcessorDetails extends PythonObjectProxy {
     /**
      * @return the type of the Processor (i.e., the class name of the Processor class)
      */

nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/processor/documentation/MultiProcessorUseCaseDetails.java

@@ -17,9 +17,11 @@
 
 package org.apache.nifi.python.processor.documentation;
 
+import org.apache.nifi.python.PythonObjectProxy;
+
 import java.util.List;
 
-public interface MultiProcessorUseCaseDetails {
+public interface MultiProcessorUseCaseDetails extends PythonObjectProxy {
     String getDescription();
 
     String getNotes();

nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/processor/documentation/ProcessorConfigurationDetails.java

@@ -17,7 +17,9 @@
 
 package org.apache.nifi.python.processor.documentation;
 
-public interface ProcessorConfigurationDetails {
+import org.apache.nifi.python.PythonObjectProxy;
+
+public interface ProcessorConfigurationDetails extends PythonObjectProxy {
     String getProcessorType();
 
     String getConfiguration();

nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/processor/documentation/PropertyDescription.java

@@ -17,7 +17,9 @@
 
 package org.apache.nifi.python.processor.documentation;
 
-public interface PropertyDescription {
+import org.apache.nifi.python.PythonObjectProxy;
+
+public interface PropertyDescription extends PythonObjectProxy {
     String getName();
 
     String getDisplayName();

nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/processor/documentation/UseCaseDetails.java

@@ -17,9 +17,11 @@
 
 package org.apache.nifi.python.processor.documentation;
 
+import org.apache.nifi.python.PythonObjectProxy;
+
 import java.util.List;
 
-public interface UseCaseDetails {
+public interface UseCaseDetails extends PythonObjectProxy {
     String getDescription();
 
     String getNotes();