[FEATURE] Add the integration with existing Spark tests on rust based server

[zuston]

Code of Conduct

• [X] I agree to follow this project's Code of Conduct

Search before asking

• [X] I have searched in the issues and found no similar issues.

Describe the feature

We should create the mini rust server to integrate with existing spark tests.

Motivation

No response

Describe the solution

No response

Additional context

No response

Are you willing to submit PR?

• [ ] Yes I am willing to submit a PR!

[guixiaowen]

@zuston i am new here, let me have a try

[zuston]

I think this is a big change. Are you interested on this? @wenlongbrother

[wenlongbrother]

Hi @zuston I am interested this task and I think it can help me have much deeper understanding of the whole project. Before start this task, here is my understanding of the task the content of this task: our goal is to create a mini rust server to successfully integrate with tests which under the integration-test dictionary. By the way, if we create rust server, do we need to keep the original Java server? Please correct me If I am wrong, Thanks.

[zuston]

I hope the mini rust based shuffle-server could be tested with the existing integration tests. But this part could be involved in the independent module rather than former intergated-test module.

[wenlongbrother]

@zuston thanks. Can you assign it to me?

[zuston]

Do you have time to do this? @guixiaowen

[wenlongbrother]

Hi @zuston do we have a default config.toml which contains basic config about shuffle server, I will use Java to start rust program. If not I will create a default one and place it in the resource directory.

[zuston]

Hi @zuston do we have a default config.toml which contains basic config about shuffle server, I will use Java to start rust program. If not I will create a default one and place it in the resource directory.

Haven't. I recommend drafting a initial implementation and then discuss it based on this.

[wenlongbrother]

Hi @zuston do we have a default config.toml which contains basic config about shuffle server, I will use Java to start rust program. If not I will create a default one and place it in the resource directory.

Haven't. I recommend drafting a initial implementation and then discuss it based on this.

ok

[wenlongbrother]

Still pending, will finish it recent days, just take examination in past serval days.

[wenlongbrother]

Hi @zuston , I am writing sendDataAndRequireBufferTest, here is a quick question why we set partition_id = 1, like below code

async fn require_buffer(
        &self,
        request: Request<RequireBufferRequest>,
    ) -> Result<Response<RequireBufferResponse>, Status> {
        let timer = GRPC_BUFFER_REQUIRE_PROCESS_TIME.start_timer();
        let req = request.into_inner();
        let app_id = req.app_id;
        let shuffle_id = req.shuffle_id;

        let app = self.app_manager_ref.get_app(&app_id);
        if app.is_none() {
            return Ok(Response::new(RequireBufferResponse {
                require_buffer_id: 0,
                status: StatusCode::NO_REGISTER.into(),
                ret_msg: "No such app in this shuffle server".to_string(),
            }));
        }

        let partition_id = PartitionedUId {
            app_id,
            shuffle_id,
            // why?
            // ignore this.
            partition_id: 1,
        };
        let app = app
            .unwrap()
            .require_buffer(RequireBufferContext {
                uid: partition_id.clone(),
                size: req.require_size as i64,
            })
            .instrument_await(format!("require buffer. uid: {:?}", &partition_id))
            .await;

        let res = match app {
            Ok(required_buffer_res) => (
                StatusCode::SUCCESS,
                required_buffer_res.ticket_id,
                "".to_string(),
            ),
            Err(err) => (StatusCode::NO_BUFFER, -1i64, format!("{:?}", err)),
        };

        timer.observe_duration();

        Ok(Response::new(RequireBufferResponse {
            require_buffer_id: res.1,
            status: res.0.into(),
            ret_msg: res.2,
        }))
    }

[zuston]

The partitionId in RequireBufferContext won't be used in the process of requiring buffer, which is only a placeholder. Please ignore this @wenlongbrother

[wenlongbrother]

Hi @zuston A follow question, I see below code will log partition_id, do you think it will confuses user?

[wenlongbrother]

Hi @zuston , can you please help review this draft pr? https://github.com/apache/incubator-uniffle/pull/1481 I am still in progress, and I wonder does this meet expectation.