fix pulsar bug

Signed-off-by: Chenlong Ma <chenlongm@vmware.com>
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: