Fix python udf source detection (#3189)

PhysicalOp relies on the input port number to determine if an operator
is a source operator. For Python UDF, from the changes in #3183, the
input ports are not correctly associated with the PhysicalOp, causing
all the Python UDFs to be recognized as source operators. This PR fixes
the issue.

core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/pythonworker/PythonProxyClient.scala

@@ -69,7 +69,7 @@ class PythonProxyClient(portNumberPromise: Promise[Int], val actorId: ActorVirtu
           logger.warn(
             s"Failed to connect to Flight Server in this attempt, retrying after $UNIT_WAIT_TIME_MS ms... remaining attempts: ${MAX_TRY_COUNT - tryCount}"
           )
-          flightClient.close()
+          if (flightClient != null) flightClient.close()
           Thread.sleep(UNIT_WAIT_TIME_MS)
           tryCount += 1
       }

core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/udf/python/PythonUDFOpDescV2.scala

@@ -93,7 +93,7 @@ class PythonUDFOpDescV2 extends LogicalOp {
       Map(operatorInfo.outputPorts.head.id -> outputSchema)
     }
 
-    if (workers > 1) {
+    val physicalOp = if (workers > 1) {
       PhysicalOp
         .oneToOnePhysicalOp(
           workflowId,
@@ -112,7 +112,10 @@ class PythonUDFOpDescV2 extends LogicalOp {
           OpExecWithCode(code, "python")
         )
         .withParallelizable(false)
-    }.withDerivePartition(_ => UnknownPartition())
+    }
+
+    physicalOp
+      .withDerivePartition(_ => UnknownPartition())
       .withInputPorts(operatorInfo.inputPorts)
       .withOutputPorts(operatorInfo.outputPorts)
       .withPartitionRequirement(partitionRequirement)