Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

CORE-20867 add a flow retry topic to handle retries of transient RPC calls #1707

Open
wants to merge 7 commits into
base: release/os/5.2
Choose a base branch
from
Open

CORE-20867 add a flow retry topic to handle retries of transient RPC calls #1707

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
13888b2
CORE-20867 add a flow retry topic to handle retries of transient RPC …
LWogan Oct 22, 2024
f7d8129
CORE-20867 add max poll records constant
LWogan Oct 22, 2024
58e9c5e
CORE-20867 add transient error retry limit timeout
LWogan Oct 22, 2024
e84724e
CORE-20867 add avro object to track sync request headers
LWogan Oct 24, 2024
45eead8
CORE-20867 simplify avro record
LWogan Oct 25, 2024
e7fd64e
CORE-20867 turn avro object into string
LWogan Oct 25, 2024
1453c99
CORE-20867 remove object not needed
LWogan Oct 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion data/avro-schema/src/main/resources/avro/net/corda/data/flow/event/FlowEvent.avsc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,8 @@
"net.corda.data.flow.event.StartFlow",
"net.corda.data.flow.output.FlowStatus",
"net.corda.data.flow.event.SessionEvent",
"net.corda.data.flow.event.external.ExternalEventResponse"
"net.corda.data.flow.event.external.ExternalEventResponse",
"net.corda.data.flow.event.external.ExternalEventRetryRequest"
]
}
]
Expand Down
13 changes: 13 additions & 0 deletions ...src/main/resources/avro/net/corda/data/flow/event/external/ExternalEventRetryRequest.avsc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
{
"type": "record",
"name": "ExternalEventRetryRequest",
"namespace": "net.corda.data.flow.event.external",
"doc": "This event captures the details of the external event request to allow it to be polled from the bus and retried.",
"fields": [
{
"name": "requestId",
"type": "string",
"doc": "The requestId of the external event request to retry."
}
]
}
2 changes: 2 additions & 0 deletions data/config-schema/src/main/java/net/corda/schema/configuration/MessagingConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ private Bus() {
public static final String KAFKA_BOOTSTRAP_SERVERS = KAFKA_PROPERTIES_COMMON + ".bootstrap.servers";
public static final String KAFKA_PROPERTIES_CONSUMER = KAFKA_PROPERTIES + ".consumer";
public static final String KAFKA_CONSUMER_MAX_POLL_INTERVAL = KAFKA_PROPERTIES_CONSUMER + ".max.poll.interval.ms";
public static final String KAFKA_CONSUMER_MAX_POLL_RECORDS = KAFKA_PROPERTIES_CONSUMER + ".max.poll.records";
public static final String KAFKA_PROPERTIES_PRODUCER = KAFKA_PROPERTIES + ".producer";
public static final String KAFKA_PRODUCER_CLIENT_ID = KAFKA_PROPERTIES_PRODUCER + ".client.id";

Expand Down Expand Up @@ -61,6 +62,7 @@ private Subscription() {
public static final String MEDIATOR_PROCESSING_MIN_POOL_RECORD_COUNT = SUBSCRIPTION + ".mediator.minPoolRecordCount";
public static final String MEDIATOR_PROCESSING_POLL_TIMEOUT = SUBSCRIPTION + ".mediator.pollTimeout";
public static final String MEDIATOR_PROCESSING_PROCESSOR_TIMEOUT = SUBSCRIPTION + ".mediator.processorTimeout";
public static final String MEDIATOR_PROCESSING_TRANSIENT_ERROR_TIMEOUT = SUBSCRIPTION + ".mediator.transientErrorRetryTimeout";
}

/**
Expand Down
7 changes: 7 additions & 0 deletions ...-schema/src/main/resources/net/corda/schema/configuration/messaging/1.0/subscription.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,13 @@
"minimum": 1000,
"maximum": 2147483647,
"default": 120000
},
"transientErrorRetryTimeout": {
"description": "The length of time in milliseconds the mediator will allow an inter-worker RPC call to return transient errors before marking the state as failed. A retry topic is used to persist retries across multiple consumer polls.",
"type": "integer",
"minimum": 1000,
"maximum": 2147483647,
"default": 1200000
}
}
}
Expand Down
2 changes: 2 additions & 0 deletions data/topic-schema/src/main/java/net/corda/schema/Schemas.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,8 @@ private Flow() {
public static final String FLOW_MAPPER_SESSION_IN = "flow.mapper.session.in";
public static final String FLOW_START = "flow.start";
public static final String FLOW_SESSION = "flow.session";

public static final String FLOW_RETRY = "flow.retry";
public static final String EXTERNAL_MESSAGE_SAMPLE = "external.message.sample";
}

Expand Down
8 changes: 8 additions & 0 deletions data/topic-schema/src/main/resources/net/corda/schema/Flow.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,4 +89,12 @@ topics:
producers:
- flow
config:
FlowRetry:
name: flow.retry
consumers:
- flow
producers:
- flow
config:


2 changes: 1 addition & 1 deletion gradle.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ cordaProductVersion = 5.2.1
# NOTE: update this each time this module contains a breaking change
## NOTE: currently this is a top level revision, so all API versions will line up, but this could be moved to
## a per module property in which case module versions can change independently.
cordaApiRevision = 53
cordaApiRevision = 54

# Main
kotlin.stdlib.default.dependency = false
Expand Down