[Core] Add Logic to Emit Task Events to Event Aggregator by MengjinYan · Pull Request #53402 · ray-project/ray

MengjinYan · 2025-05-29T15:09:08Z

Why are these changes needed?

This PR adds the logic to emit task events to event aggregator, mainly:

Update the task event buffer logic to convert and send the ray event to the event aggregator at the same time when the task events are converted and sent to GCS
Added 2 configs to control turning on/off the path to send to GCS and the path to send to event aggregator
Added test task event buffer tests for the ray event to aggregator path

Related issue number

N/A

Checks

I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
I've run scripts/format.sh to lint the changes in this PR.
I've included any doc changes needed for https://docs.ray.io/en/master/.
- I've added any new APIs to the API Reference. For example, if I added a
  method in Tune, I've added it in doc/source/tune/api/ under the
  corresponding .rst file.
I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
Testing Strategy
- Unit tests
- Release tests
- This PR is not tested :(

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

src/ray/core_worker/task_event_buffer.h

src/ray/core_worker/task_event_buffer.cc

src/ray/core_worker/event_aggregator_exporter.cc

dayshah · 2025-06-03T21:08:30Z

src/ray/core_worker/task_event_buffer.cc

+  if (send_ray_events_to_aggregator_enabled_) {
+    RAY_CHECK(data->ray_event_data);
+    SendRayEventsToAggregator(std::move(*data->ray_event_data));
+  }


should it be possible to send to gcs and aggregator?

The plan is to move the event -> gcs path to the event aggregator. So here we just need to support event sending to the event aggregator and later we will need to add the logic for the aggregator to send to gcs.

we could segfault if both are enabled because both move the same thing

I think one moves the task_event_data and the other moves the ray_event_data. And they are created separately. So I think we can have them enabled at the same time without interfere with each other. But let me know if I missed anything.

src/ray/core_worker/task_event_buffer.cc

Signed-off-by: myan <myan@anyscale.com>

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

israbbani

I did a first pass. I'll do another pass tomorrow morning!

israbbani · 2025-06-25T01:04:55Z

src/ray/core_worker/core_worker.cc

-      std::make_shared<gcs::GcsClient>(options_.gcs_options));
+      std::make_shared<gcs::GcsClient>(options_.gcs_options),
+      std::make_unique<rpc::EventAggregatorClientImpl>(
+          "127.0.0.1", options_.metrics_agent_port, *client_call_manager_));


Is there ever a remote EventAggregatorClient? If not, we don't need the address as a parameter and can add it later once it's needed.

Fixed in df01b29

src/ray/common/grpc_util.h

israbbani · 2025-06-25T22:57:29Z

src/ray/common/ray_config_def.h

+// Whether the task events from the core worker are sent to GCS directly.
+RAY_CONFIG(bool, enable_core_worker_task_event_to_gcs, true)
+
+// Whether to enable the ray event to send to the event aggregator.
+// Currently, only task events are supported.
+RAY_CONFIG(bool, enable_core_worker_ray_event_to_aggregator, false)


Will these be removed after the migration is done? Can we add an issue/todo to make sure these are cleaned up afterwards?

We should have a separate follow up (team-wide) to move feature flags to a separate config file so it's obvious to users that they should not rely on them.

Fixed in df01b29 and e769c90

israbbani · 2025-06-25T23:03:29Z

src/ray/core_worker/core_worker.cc

        /* attempt_number */ 0,
        rpc::TaskStatus::FINISHED,
-        /* timestamp */ absl::GetCurrentTimeNanos());
+        /* timestamp */ absl::GetCurrentTimeNanos(),


Suggested change

/* timestamp */ absl::GetCurrentTimeNanos(),

/*timestamp=*/ absl::GetCurrentTimeNanos(),

I don't like this convention, but we use it elsewhere so this is just to make it consistent.

Updated in df01b29

src/ray/core_worker/event_aggregator_exporter.h

src/ray/core_worker/task_event_buffer.h

BUILD.bazel

src/ray/core_worker/event_aggregator_exporter.h

israbbani · 2025-06-26T01:35:52Z

src/ray/core_worker/task_event_buffer.cc

+
+  if (is_actor_task_event_) {
+    if (is_definition_event) {
+      ray_event.set_message("Actor task definition event");


What's the point of having the message when there's an enum?

This is a general field mainly to put high level string information about the event. I might not be applicable for task events. I'll just not set it to avoid unnecessary information sent.

Updated in df01b29

src/ray/core_worker/task_event_buffer.cc

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

MengjinYan · 2025-07-16T01:41:38Z

@israbbani I should fixed all the comments that you have and it should be ready for another round of review

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

can-anyscale

qq, do we have an e2e python test that check the aggregator has the events we expect, etc.?

can-anyscale · 2025-07-03T17:50:47Z

src/ray/core_worker/task_event_buffer.cc

+  if (state_update_->node_id_.has_value()) {
+    RAY_CHECK(task_status_ == rpc::TaskStatus::SUBMITTED_TO_WORKER)
+            .WithField("TaskStatus", task_status_)
+        << "Node ID should be included when task status changes to "


the check is the inverse of this comment though, "Task status must be SUBMITTED_TO_WORKER if node ID is included"

can-anyscale · 2025-07-03T17:50:53Z

src/ray/core_worker/task_event_buffer.cc

+  }
+
+  if (state_update_->worker_id_.has_value()) {
+    RAY_CHECK(task_status_ == rpc::TaskStatus::SUBMITTED_TO_WORKER)


can-anyscale · 2025-07-03T17:53:44Z

src/ray/core_worker/task_event_buffer.cc

+}
+
+void TaskStatusEvent::ToRpcRayEvents(
+    std::pair<std::optional<rpc::events::RayEvent>, std::optional<rpc::events::RayEvent>>


nit: maybe define a type for this, something like
using TaskDefinitionAndExecutionEvents = std::pair<std::optional<rpc::events::RayEvent>, std::optional<rpc::events::RayEvent>>

dayshah · 2025-07-22T06:39:43Z

src/mock/ray/rpc/event_aggregator_client.h

+              (std::unique_ptr<rpc::events::RayEventsData> data_ptr,
+               std::function<void(Status status)> callback),
+              (override));
+};


i think we want to move away from using the shared mock folder where everything inside it is mocked, would want to just enclose this inside the task_event_buffer_test so other uses are encouraged to use fake implementations instead of a mock

dayshah · 2025-07-22T06:41:46Z

src/ray/core_worker/core_worker.cc

          /*attempt_number=*/0,
          rpc::TaskStatus::RUNNING,
          /*timestamp=*/absl::GetCurrentTimeNanos(),
+          is_actor_task_event,


Suggested change

is_actor_task_event,

/*is_actor_task_event=*/spec.IsActorTask(),

personal nit but inline if just used in one spot

dayshah · 2025-07-22T06:48:31Z

src/ray/core_worker/task_event_buffer.cc

+  auto task_state = execution_event_data.mutable_task_state();
+  if (task_status_ != rpc::TaskStatus::NIL) {
+    (*task_state)[task_status_] = timestamp;


Suggested change

auto task_state = execution_event_data.mutable_task_state();

if (task_status_ != rpc::TaskStatus::NIL) {

(*task_state)[task_status_] = timestamp;

auto &task_state = *execution_event_data.mutable_task_state();

if (task_status_ != rpc::TaskStatus::NIL) {

task_state[task_status_] = timestamp;

nit, a little cleaner

dayshah · 2025-07-22T06:53:01Z

src/ray/core_worker/task_event_buffer.cc

+          auto [itr_ray_events, _] = agg_ray_events.try_emplace(
+              event->GetTaskAttempt(),
+              std::pair<std::optional<rpc::events::RayEvent>,
+                        std::optional<rpc::events::RayEvent>>());


Suggested change

auto [itr_ray_events, _] = agg_ray_events.try_emplace(

event->GetTaskAttempt(),

std::pair<std::optional<rpc::events::RayEvent>,

std::optional<rpc::events::RayEvent>>());

auto [itr_ray_events, _] = agg_ray_events.try_emplace(

event->GetTaskAttempt());

dayshah · 2025-07-22T06:54:09Z

src/ray/core_worker/task_event_buffer.cc

+        if (send_task_events_to_gcs_enabled_) {
+          auto [itr_task_events, _] =
+              agg_task_events.try_emplace(event->GetTaskAttempt(), rpc::TaskEvents());
+          event->ToRpcTaskEvents(&(itr_task_events->second));


Suggested change

event->ToRpcTaskEvents(&(itr_task_events->second));

agg_task_events.try_emplace(event->GetTaskAttempt());

event->ToRpcTaskEvents(&itr_task_events->second);

also ToRpcRayEvents takes by ref, this one takes by ptr, any reason for inconsistency?

No particular reason that I'm aware of. This is legacy code and we will remove the code path after the task events fully migrated to the event aggregator.

dayshah · 2025-07-22T07:02:08Z

src/ray/core_worker/task_event_buffer.cc

+  if (send_ray_events_to_aggregator_enabled_) {
+    RAY_CHECK(data->ray_event_data);
+    SendRayEventsToAggregator(std::move(*data->ray_event_data));
+  }


we could segfault if both are enabled because both move the same thing

dayshah · 2025-07-22T07:09:04Z

src/ray/rpc/event_aggregator_client.h

+              });
+
+    return Status::OK();
+  }


this function doesn't really seem to be adding much on top of AddEvents imo and is kind of a downgrade from the standard implementation for these 3 reasons.

The debug is hidden pretty deep within the stack

Always returning Status::OK so returning status doesn't help

Forcing the caller to pass in a unique ptr kills the optimization opportunity that could be had by moving RayEventsData in.

dayshah · 2025-07-22T07:09:23Z

src/ray/rpc/event_aggregator_client.h

+  }
+
+ private:
+  void AddEvents(const rpc::events::AddEventRequest &request,


why not use the macro we use in the rest of the codebase to stay consistent

dayshah · 2025-07-22T07:11:15Z

src/ray/rpc/event_aggregator_client.h

        request,
        callback,
        "EventAggregatorService.grpc_client.AddEvents",
        // TODO(myan): Add timeout and retry logic.


this should be in the proto file, or should be handled in this PR. Based on above, I think it doesn't matter that much if it fails. Should just state that in the proto file the way it is in core_worker.proto and node_manager.proto

dayshah · 2025-07-22T07:13:40Z

src/ray/core_worker/task_event_buffer.cc

+    std::pair<std::optional<rpc::events::RayEvent>, std::optional<rpc::events::RayEvent>>
+        &ray_events) {
+  google::protobuf::Timestamp timestamp = AbslTimeNanosToProtoTimestamp(timestamp_);
+


can you add an auto &[name1, name2] = ray_events; up here, more readable than using .first, .second throughout

MengjinYan · 2025-07-23T21:29:31Z

qq, do we have an e2e python test that check the aggregator has the events we expect, etc.?

@can-anyscale I tested manually for now to not make the PR even larger. And I'm planning to add the integration tests in a followup PR.

The manual test result is as follows:

Test Ray script: https://gist.github.com/MengjinYan/031372dfb244c13f4905cfdce32cb388
Command to run the script:

# start ray
RAY_enable_core_worker_ray_event_to_aggregator=1 ray start --head
# run the script 
python test-tasks.py

The events I got from a local http echo server: https://gist.github.com/MengjinYan/b8bc27086bad6ba220d452fb221d7111

can-anyscale

looks good on my end for the event exporting logic, defer to others for other parts, thanks

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

MengjinYan · 2025-07-25T21:20:20Z

@can-anyscale's and @dayshah's comments.

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

israbbani

Reviewd everything except for the task_event_buffer_test. Looks good. Left some nits and a few clarification questions.

israbbani · 2025-07-28T16:24:26Z

BUILD.bazel

    ],
 )

+# Event Aggregator gRPC lib.


I don't think the comment adds anything.

israbbani · 2025-07-28T16:42:44Z

src/ray/core_worker/core_worker.cc

+        /*attempt_number=*/0,
        rpc::TaskStatus::FINISHED,
-        /* timestamp */ absl::GetCurrentTimeNanos());
+        /*timestamp=*/absl::GetCurrentTimeNanos(),


Are these automatically generated by a linter?

No, I just added them.

israbbani · 2025-07-28T16:48:20Z

src/ray/core_worker/task_event_buffer.cc

+  definition_event_data.mutable_ref_ids()->insert(labels.begin(), labels.end());
+
+  // Specific fields
+  if constexpr (std::is_same_v<T, rpc::events::ActorTaskDefinitionEvent>) {


Sounds good. Just to check my understanding, once the migration is complete, this function will no longer need a template type?

israbbani · 2025-07-28T21:11:41Z

src/ray/core_worker/task_event_buffer.h

+  /// \param[out] ray_events The pair of TaskDefinitionEvent and TaskExecutionEvent to be
+  /// filled.


Might be helpful for the reader if you clarified that TaskDefinitionEvent and TaskExecutionEvent are both proto messages.

israbbani · 2025-07-28T21:31:17Z

src/ray/core_worker/task_event_buffer.h

  void ToRpcTaskExportEvents(
      std::shared_ptr<rpc::ExportTaskEventData> rpc_task_export_event_data) override;

+  void ToRpcRayEvents(RayEventsPair &ray_events) override;


Need to add documentation for the public API.

israbbani · 2025-07-29T01:06:44Z

src/ray/protobuf/events_event_aggregator_service.proto

 }

-message AddEventRequest {
+message AddEventsRequest {


Why did everything get renamed to Events instead of Event?

There are some naming inconsistency with the previous implementation of the event aggregator so I make everything consistent with the same naming convention.

israbbani · 2025-07-29T01:07:08Z

src/ray/protobuf/events_event_aggregator_service.proto

-  rpc AddEvents(AddEventRequest) returns (AddEventReply);
+  // Add events to the local event aggregator.
+  // Failure:
+  // Infinite timeout because the communcaiton to the event aggregator will


Suggested change

// Infinite timeout because the communcaiton to the event aggregator will

// Infinite timeout because the communication to the event aggregator will

israbbani · 2025-07-29T01:08:18Z

src/ray/core_worker/test/task_event_buffer_test.cc

+                                const rpc::ClientCallback<rpc::events::AddEventsReply> &>
+                   &args) override {
+    auto &request = std::get<0>(args);
+    RAY_LOG(INFO) << "[myan] request.events_data().events_size()="


Probably left in from debugging.

israbbani · 2025-07-29T01:09:36Z

src/ray/core_worker/test/task_event_buffer_test.cc

+                                   const rpc::events::RayEventsData &expect_data) {
+    RAY_LOG(INFO) << "[myan] actual_data.events_size()=" << actual_data.events_size();
+
+    // Sore and compare


Suggested change

// Sore and compare

// Store and compare

israbbani · 2025-07-29T01:09:37Z

src/ray/core_worker/test/task_event_buffer_test.cc


+  static void CompareRayEventsData(const rpc::events::RayEventsData &actual_data,
+                                   const rpc::events::RayEventsData &expect_data) {
+    RAY_LOG(INFO) << "[myan] actual_data.events_size()=" << actual_data.events_size();


Same as above

israbbani

Tests look okay to me. Leaving feedback for future work for when we refactor this

We should refactor out the event pushing logic from the TaskEventBuffer. The public API will become pretty simple (AddEvent, GetEvents) and that will make writing tests against the public API better.
For the tests themselves, I use the convention UnitOfWorkConditionUnderTestExpectResult for naming. GTest doesn't let you use underscores, but we can standardize on a new delimiter. E.g. AddEventDropsEventsWhenBufferIsFull or something like that.

Overall, looks good and ready to merge once nits have been addressed.

edoakes · 2025-07-30T00:53:06Z

python/ray/dashboard/modules/aggregator/aggregator_agent.py

                    f", and {len(error_messages) - truncate_num} more events dropped"
                )
-        status = events_event_aggregator_service_pb2.AddEventStatus(
+        status = events_event_aggregator_service_pb2.AddEventsStatus(


why do we need a custom AddEventsStatus -- are gRPC status codes & messages insufficient?

The status in the grpc response is mainly to indicate whether all events in the batch are successfully added to the event aggregator, while I think the GRPC status is mainly indicate whether the response is successfully received.

I'm actually planning to have a followup PR to improve the response to return the number of events failed to add to the event aggregator due to unknown errors and also track the number events dropped due to a full event aggregator on the aggregator side. And probably remove the status & error message.

What action will the client take based on the fine-grained status information such as the count of events that were failed to be added?

The count is mainly for observability purposes so that we can have metrics to understand how many events are missing and at which place they are dropped and why.

this seems like premature optimization/over-engineering to me. couldn't the server itself export the metrics about the number of dropped events? and why would we expect that partial drops will be a common occurrence (vs. all events being recorded or all being dropped)

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

Signed-off-by: myan <myan@anyscale.com>

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

…#53402) This PR adds the logic to emit task events to event aggregator, mainly: * Update the task event buffer logic to convert and send the ray event to the event aggregator at the same time when the task events are converted and sent to GCS * Added 2 configs to control turning on/off the path to send to GCS and the path to send to event aggregator * Added test task event buffer tests for the ray event to aggregator path --------- Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com> Signed-off-by: myan <myan@anyscale.com> Signed-off-by: avigyabb <avigyabb@stanford.edu>

This PR adds the logic to emit task events to event aggregator, mainly: * Update the task event buffer logic to convert and send the ray event to the event aggregator at the same time when the task events are converted and sent to GCS * Added 2 configs to control turning on/off the path to send to GCS and the path to send to event aggregator * Added test task event buffer tests for the ray event to aggregator path --------- Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com> Signed-off-by: myan <myan@anyscale.com>

This PR adds the logic to emit task events to event aggregator, mainly: * Update the task event buffer logic to convert and send the ray event to the event aggregator at the same time when the task events are converted and sent to GCS * Added 2 configs to control turning on/off the path to send to GCS and the path to send to event aggregator * Added test task event buffer tests for the ray event to aggregator path --------- Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com> Signed-off-by: myan <myan@anyscale.com> Signed-off-by: Kamil Kaczmarek <kamil@anyscale.com>

…#53402) This PR adds the logic to emit task events to event aggregator, mainly: * Update the task event buffer logic to convert and send the ray event to the event aggregator at the same time when the task events are converted and sent to GCS * Added 2 configs to control turning on/off the path to send to GCS and the path to send to event aggregator * Added test task event buffer tests for the ray event to aggregator path --------- Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com> Signed-off-by: myan <myan@anyscale.com> Signed-off-by: Michael Acar <michael.j.acar@gmail.com>

…#53402) This PR adds the logic to emit task events to event aggregator, mainly: * Update the task event buffer logic to convert and send the ray event to the event aggregator at the same time when the task events are converted and sent to GCS * Added 2 configs to control turning on/off the path to send to GCS and the path to send to event aggregator * Added test task event buffer tests for the ray event to aggregator path --------- Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com> Signed-off-by: myan <myan@anyscale.com> Signed-off-by: jugalshah291 <shah.jugal291@gmail.com>

…#53402) This PR adds the logic to emit task events to event aggregator, mainly: * Update the task event buffer logic to convert and send the ray event to the event aggregator at the same time when the task events are converted and sent to GCS * Added 2 configs to control turning on/off the path to send to GCS and the path to send to event aggregator * Added test task event buffer tests for the ray event to aggregator path --------- Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com> Signed-off-by: myan <myan@anyscale.com> Signed-off-by: Douglas Strodtman <douglas@anyscale.com>

MengjinYan added 7 commits May 29, 2025 03:32

add logic to emit ray task events to event aggregator

1c41557

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

fix todo and add more tests

1559baf

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

Merge branch 'master' into core-703-3

3666e2d

fix test failure

ca9d33a

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

Merge branch 'master' into core-703-3

09cb8b8

further fix tests

e372eec

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

Merge branch 'master' into core-703-3

0791f78

dayshah reviewed Jun 3, 2025

View reviewed changes

MengjinYan and others added 2 commits June 5, 2025 00:27

fix test

0d4c835

Signed-off-by: myan <myan@anyscale.com>

Merge branch 'master' into core-703-3

821b2a0

MengjinYan added the go add ONLY when ready to merge, run all tests label Jun 5, 2025

MengjinYan marked this pull request as ready for review June 9, 2025 18:49

MengjinYan added 3 commits June 18, 2025 22:48

fix comment

b9ee5db

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

Merge branch 'master' into core-703-3

c58e8ef

fix test failure

7f09590

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

edoakes assigned israbbani Jun 25, 2025

israbbani reviewed Jun 26, 2025

View reviewed changes

MengjinYan requested a review from can-anyscale July 2, 2025 21:48

MengjinYan added 3 commits July 15, 2025 10:45

fix comment

df01b29

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

Merge branch 'master' into core-703-3

7d106b4

fix comment

e769c90

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

MengjinYan added 2 commits July 16, 2025 09:59

Merge branch 'master' into core-703-3

71fb238

fix test failure

189726d

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

MengjinYan requested a review from a team as a code owner July 17, 2025 06:16

MengjinYan added 2 commits July 16, 2025 23:18

Merge branch 'master' into core-703-3

da230fc

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

Merge branch 'master' into core-703-3

e9457cf

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

jjyao requested review from dayshah and israbbani July 18, 2025 17:56

fix test failure

65ef8ad

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

can-anyscale reviewed Jul 22, 2025

View reviewed changes

dayshah reviewed Jul 22, 2025

View reviewed changes

can-anyscale approved these changes Jul 23, 2025

View reviewed changes

fix review comment

25a74fa

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

MengjinYan added 2 commits July 28, 2025 10:58

Merge branch 'master' into core-703-3

43d98ef

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

fix the task_event_buffer_export_event_test failure

0d9ccfe

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

israbbani reviewed Jul 29, 2025

View reviewed changes

israbbani approved these changes Jul 29, 2025

View reviewed changes

Merge branch 'master' into core-703-3

796ca6c

edoakes reviewed Jul 30, 2025

View reviewed changes

MengjinYan and others added 5 commits July 29, 2025 22:45

Merge branch 'master' into core-703-3

81d1284

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

fix comment & merge conflict

772e4b8

Signed-off-by: myan <myan@anyscale.com>

fix test failure

c72888d

Signed-off-by: myan <myan@anyscale.com>

Merge branch 'master' into core-703-3

696bce6

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

fix build failure

d966b0b

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

edoakes enabled auto-merge (squash) July 31, 2025 21:56

edoakes merged commit bbbe9ef into ray-project:master Jul 31, 2025
6 checks passed

	/* timestamp */ absl::GetCurrentTimeNanos(),
	/timestamp=/ absl::GetCurrentTimeNanos(),

	is_actor_task_event,
	/is_actor_task_event=/spec.IsActorTask(),

	event->ToRpcTaskEvents(&(itr_task_events->second));
	agg_task_events.try_emplace(event->GetTaskAttempt());
	event->ToRpcTaskEvents(&itr_task_events->second);

		/// \param[out] ray_events The pair of TaskDefinitionEvent and TaskExecutionEvent to be
		/// filled.

	// Infinite timeout because the communcaiton to the event aggregator will
	// Infinite timeout because the communication to the event aggregator will

Conversation

MengjinYan commented May 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Why are these changes needed?

Related issue number

Checks

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

israbbani left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

MengjinYan commented Jul 16, 2025

Uh oh!

can-anyscale left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

MengjinYan commented May 29, 2025 •

edited

Loading