fix pulsar bug

Signed-off-by: Chenlong Ma <[email protected]>
FederatedAI · Nov 24, 2023 · 8c9db8a · 8c9db8a
1 parent 86c1b73
commit 8c9db8a
Show file tree

Hide file tree

Showing 12 changed files with 35 additions and 32 deletions.
diff --git a/docker-deploy/generate_config.sh b/docker-deploy/generate_config.sh
@@ -206,11 +206,11 @@ GenerateConfig() {
 				if [ "$federation" == "RabbitMQ" ]; then
 					cp -r training_template/backends/spark/rabbitmq confs-$party_id/confs/
 					# delete Pulsar spec
-					sed -i '202,216d' confs-"$party_id"/docker-compose.yml
+					sed -i '203,217d' confs-"$party_id"/docker-compose.yml
 				elif [ "$federation" == "Pulsar" ]; then
 					cp -r training_template/backends/spark/pulsar confs-$party_id/confs/
 					# delete RabbitMQ spec
-					sed -i '183,200d' confs-"$party_id"/docker-compose.yml
+					sed -i '184,201d' confs-"$party_id"/docker-compose.yml
 				fi
 			fi
 		fi
@@ -225,10 +225,10 @@ GenerateConfig() {
 				# federation
 				if [ "$federation" == "RabbitMQ" ]; then
 					cp -r training_template/backends/spark/rabbitmq confs-$party_id/confs/
-					sed -i '146,160d' confs-$party_id/docker-compose.yml
+					sed -i '149,163d' confs-$party_id/docker-compose.yml
 				elif [ "$federation" == "Pulsar" ]; then
 					cp -r training_template/backends/spark/pulsar confs-$party_id/confs/
-					sed -i '128,144d' confs-$party_id/docker-compose.yml
+					sed -i '131,147d' confs-$party_id/docker-compose.yml
 				fi
 			fi
 		fi
@@ -512,11 +512,7 @@ ${party_id}:
     port: 6650
     sslPort: 6651
     proxy: ""
-default:
-  proxy: "proxy.fate.org:443"
-  domain: "fate.org"
-  brokerPort: 6650
-  brokerSslPort: 6651
+
 EOF
 
 		fi

diff --git a/docker-deploy/training_template/docker-compose-spark-slim.yml b/docker-deploy/training_template/docker-compose-spark-slim.yml
@@ -87,6 +87,9 @@ services:
     - "-c"
     - |
         set -x
+        sed -i "s/int(party.party_id)/str(party.party_id)/g" /data/projects/fate/fate/python/fate/arch/federation/pulsar/_federation.py
+        cp /data/projects/fate/fate_flow/conf/pulsar_route_table.yaml /data/projects/fate/fate_flow/pulsar_route_table.yaml
+        cp /data/projects/fate/fate_flow/conf/rabbitmq_route_table.yaml /data/projects/fate/fate_flow/rabbitmq_route_table.yaml
         sleep 5 && python fateflow/python/fate_flow/fate_flow_server.py
     environment:
       FATE_PROJECT_BASE: "/data/projects/fate"
@@ -144,14 +147,14 @@ services:
       - fate-network
 
   pulsar:
-    image: "federatedai/pulsar:2.7.0"
+    image: "federatedai/pulsar:2.10.2"
     ports:
       - "6650:6650"
       - "6651:6651"
       - "8001:8080"
     volumes:
       - ./confs/pulsar/standalone.conf:/pulsar/conf/standalone.conf
-      - ./shared_dir/data/pulsar:/pulsar/data
+      # - ./shared_dir/data/pulsar:/pulsar/data
       - /etc/localtime:/etc/localtime:ro
     command:
       ["/bin/bash", "-c", "bin/pulsar standalone -nss"]

diff --git a/docker-deploy/training_template/docker-compose-spark.yml b/docker-deploy/training_template/docker-compose-spark.yml
@@ -87,6 +87,7 @@ services:
     - "-c"
     - |
         set -x
+        sed -i "s/int(party.party_id)/str(party.party_id)/g" /data/projects/fate/fate/python/fate/arch/federation/pulsar/_federation.py
         cp /data/projects/fate/fate_flow/conf/pulsar_route_table.yaml /data/projects/fate/fate_flow/pulsar_route_table.yaml
         cp /data/projects/fate/fate_flow/conf/rabbitmq_route_table.yaml /data/projects/fate/fate_flow/rabbitmq_route_table.yaml
         sleep 5 && python fate_flow/python/fate_flow/fate_flow_server.py
@@ -205,9 +206,10 @@ services:
       - "6650:6650"
       - "6651:6651"
       - "8001:8080"
+    user: root
     volumes:
       - ./confs/pulsar/standalone.conf:/pulsar/conf/standalone.conf
-      - ./shared_dir/data/pulsar:/pulsar/data
+      # - ./shared_dir/data/pulsar:/pulsar/data
       - /etc/localtime:/etc/localtime:ro
     command:
       ["/bin/bash", "-c", "bin/pulsar standalone -nss"]

diff --git a/docs/FATE_On_Spark_With_Pulsar.md b/docs/FATE_On_Spark_With_Pulsar.md
@@ -2,7 +2,7 @@
 
 ## Overview
 
-FATE supports using [Spark](https://spark.apache.org/) as a computing engine since v1.5.0 Along with Spark, it also requires HDFS and RabbitMQ as storage and transmission service respectively, to compose a functional FATE cluster. In v1.6.0, the FATE also supports to use [Pulsar](https://pulsar.apache.org/admin-rest-api/?version=2.7.0&apiversion=v2#tag/clusters) as the transmission engine, a user can switch the transmission engine easily. Ideally, the Pulsar provides better throughput and scalability, more importantly, organizations can compose FATE clusters of star network using Pulsar. The overall architecture of "FATE on Spark with Pulsar" is as the following diagram:
+FATE supports using [Spark](https://spark.apache.org/) as a computing engine since v1.5.0 Along with Spark, it also requires HDFS and RabbitMQ as storage and transmission service respectively, to compose a functional FATE cluster. In v1.6.0, the FATE also supports to use [Pulsar](https://pulsar.apache.org/admin-rest-api/?version=2.10.2&apiversion=v2#tag/clusters) as the transmission engine, a user can switch the transmission engine easily. Ideally, the Pulsar provides better throughput and scalability, more importantly, organizations can compose FATE clusters of star network using Pulsar. The overall architecture of "FATE on Spark with Pulsar" is as the following diagram:
 
 <div style="text-align:center", align=center>
 <img src="./images/fate_on_spark_with_pulsar.png" />
@@ -86,7 +86,7 @@ When submitting a task, the user can declare in the config file to use Pulsar as
    }
 ```
 
-Generally, there is no need to set such a configuration. As for the available parameters, please refer to the [`create_producer`](https://pulsar.apache.org/api/python/2.7.0-SNAPSHOT/#pulsar.Client.create_producer) and [`subscribe`](https://pulsar.apache.org/api/python/2.7.0-SNAPSHOT/#pulsar.Client.subscribe) methods in the Pulsar python client.
+Generally, there is no need to set such a configuration. As for the available parameters, please refer to the [`create_producer`](https://pulsar.apache.org/api/python/2.10.2-SNAPSHOT/#pulsar.Client.create_producer) and [`subscribe`](https://pulsar.apache.org/api/python/2.10.2-SNAPSHOT/#pulsar.Client.subscribe) methods in the Pulsar python client.
 
 ## Deployment of Star Network
 

diff --git a/helm-charts/FATE/templates/core/fateflow/configmap.yaml b/helm-charts/FATE/templates/core/fateflow/configmap.yaml
@@ -203,6 +203,18 @@ metadata:
 {{ include "fate.labels" . | indent 4 }}
 data:
   pulsar_route_table.yaml: |
+    {{- if .Values.modules.pulsar.route_table }}
+    {{- range $key, $val := .Values.modules.pulsar.route_table }}
+    {{ $key }}:
+{{ toYaml . | indent 6 }}
+    {{- end }}
+    {{- else }}
+    {{ .Values.partyId }}:
+      host: pulsar
+      port: 6650
+      sslPort: 6651
+      proxy: ""
+    {{- end}}
     {{- with .Values.modules.pulsar.exchange }}
     default:
       proxy: "{{ .ip }}:{{ .port }}"
@@ -214,18 +226,6 @@ data:
       brokerPort: 6650
       brokerSslPort: 6651
     {{- end }}
-  {{- if .Values.modules.pulsar.route_table }}
-    {{- range $key, $val := .Values.modules.pulsar.route_table }}
-    {{ $key }}:
-{{ toYaml . | indent 6 }}
-    {{- end }}
-  {{- else }}
-    {{ .Values.partyId }}:
-      host: pulsar
-      port: 6650
-      sslPort: 6651
-      proxy: ""
-  {{- end}}
 ---
 kind: ConfigMap
 apiVersion: v1

diff --git a/helm-charts/FATE/templates/core/python-spark.yaml b/helm-charts/FATE/templates/core/python-spark.yaml
@@ -121,6 +121,8 @@ spec:
                 cp /data/projects/spark-3.1.3-bin-hadoop3.2/conf/spark-defaults-template.conf /data/projects/spark-3.1.3-bin-hadoop3.2/conf/spark-defaults.conf
                 sed -i "s/fateflow/${POD_IP}/g" /data/projects/spark-3.1.3-bin-hadoop3.2/conf/spark-defaults.conf
                 
+                sed -i "s/int(party.party_id)/str(party.party_id)/g" /data/projects/fate/fate/python/fate/arch/federation/pulsar/_federation.py
+
                 cp /data/projects/fate/fate_flow/conf/pulsar_route_table/pulsar_route_table.yaml /data/projects/fate/fate_flow/pulsar_route_table.yaml
                 cp /data/projects/fate/fate_flow/conf/rabbitmq_route_table/rabbitmq_route_table.yaml /data/projects/fate/fate_flow/rabbitmq_route_table.yaml
         

diff --git a/helm-charts/FATE/values-template-example.yaml b/helm-charts/FATE/values-template-example.yaml
@@ -399,7 +399,7 @@ skippedKeys:
 
 # pulsar:
   # image: "federatedai/pulsar"
-  # imageTag: "2.7.0"
+  # imageTag: "2.10.2"
 #   nodeSelector:
 #   tolerations:
 #   affinity:

diff --git a/helm-charts/FATE/values-template.yaml b/helm-charts/FATE/values-template.yaml
@@ -650,7 +650,7 @@ modules:
     {{- end }}
     type: {{ .type | default "ClusterIP" }}
     image: {{ .image | default "federatedai/pulsar" | quote }}
-    imageTag: {{ .imageTag | default "2.7.0" | quote }}
+    imageTag: {{ .imageTag | default "2.10.2" | quote }}
     httpNodePort: {{ .httpNodePort }}
     httpsNodePort: {{ .httpsNodePort }}
     loadBalancerIP: {{ .loadBalancerIP }}

diff --git a/helm-charts/FATE/values.yaml b/helm-charts/FATE/values.yaml
@@ -366,7 +366,7 @@ modules:
   pulsar:
     include: true
     image: "federatedai/pulsar"
-    imageTag: "2.7.0"
+    imageTag: "2.10.2"
     nodeSelector:
     tolerations:
     env:

diff --git a/helm-charts/Images_list.md b/helm-charts/Images_list.md
@@ -16,7 +16,7 @@
 - nginx:1.17
 - federatedai/nginx:${version-tag}
 - federatedai/rabbitmq:3.8.3-management
-- federatedai/pulsar:2.7.0
+- federatedai/pulsar:2.10.2
 
 ## FATE-Serving
 

diff --git a/k8s-deploy/cluster-spark-pulsar.yaml b/k8s-deploy/cluster-spark-pulsar.yaml
@@ -253,7 +253,7 @@ skippedKeys:
 
 # pulsar:
   # image: "federatedai/pulsar"
-  # imageTag: "2.7.0"
+  # imageTag: "2.10.2"
 #   nodeSelector:
 #   tolerations:
 #   affinity:

diff --git a/k8s-deploy/cluster-spark-slim.yaml b/k8s-deploy/cluster-spark-slim.yaml
@@ -183,7 +183,7 @@ skippedKeys:
           # grpc_port: 30102
 # pulsar:
   # image: "federatedai/pulsar"
-  # imageTag: "2.7.0"
+  # imageTag: "2.10.2"
 #   nodeSelector:
 #   tolerations:
 #   affinity: