JinkunLiu commented on code in PR #950:
URL: https://github.com/apache/flink-agents/pull/950#discussion_r3944593524
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java:
##########
@@ -144,6 +145,52 @@ void testExecuteAgent() throws Exception {
}
}
+ @Test
+ void testJavaEventAttachmentsAreOffloadedAndResolvedBetweenActions()
throws Exception {
+ InMemoryActionStateStore actionStateStore = new
InMemoryActionStateStore(false);
+ AgentPlan agentPlan = TestAgent.getEventAttachmentAgentPlan();
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory<>(agentPlan, true,
actionStateStore),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class))) {
+ testHarness.open();
+ ActionExecutionOperator<Long, Object> operator =
+ (ActionExecutionOperator<Long, Object>)
testHarness.getOperator();
+
+ List<Event> eventsAtSendBoundary = new ArrayList<>();
+ operator.getEventRouter()
+ .addEventListener(
+ (context, event) -> {
+ if
(TestAgent.ATTACHMENT_EVENT_TYPE.equals(event.getType())) {
+ eventsAtSendBoundary.add(event);
+ }
+ });
+
+ testHarness.processElement(new StreamRecord<>(1L));
+ operator.waitInFlightEventsFinished();
+
+ assertThat(eventsAtSendBoundary).hasSize(1);
+ Event runtimeEvent = eventsAtSendBoundary.get(0);
+ Object reference =
runtimeEvent.getAttachment(TestAgent.ATTACHMENT_KEY);
+ assertThat(reference).isInstanceOf(MemoryRef.class);
+ List<StreamRecord<Object>> recordOutput =
+ (List<StreamRecord<Object>>) testHarness.getRecordOutput();
+ assertThat(recordOutput).hasSize(1);
+
assertThat(recordOutput.get(0).getValue()).isEqualTo(Map.of("value", 1L));
+
+ ActionState actionState =
+ actionStateStore.get(
+ 1L,
+ 0L,
+
agentPlan.getActions().get("receiveEventAttachment"),
+ runtimeEvent);
+ assertThat(actionState).isNotNull();
+
assertThat(actionState.getTaskEvent().getAttachment(TestAgent.ATTACHMENT_KEY))
+ .isSameAs(reference);
Review Comment:
Good catch, thanks. Since the existing assertion already covers the intended
behavior, I think this assertion can simply be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
