Enable event-driven function retries

constfunctions=require('@google-cloud/functions-framework');
/**
 * Cloud Event Function that only executes within
 * a certain time period after the triggering event
 *
 * @param {object} event The Cloud Functions event.
 * @param {function} callback The callback function.
 */
functions.cloudEvent('avoidInfiniteRetries',(event,callback)=>{
consteventAge=Date.now()-Date.parse(event.time);
consteventMaxAge=10000;
// Ignore events that are too old
if(eventAge > eventMaxAge){
console.log(`Dropping event ${event} with age ${eventAge} ms.`);
callback();
return;
}
// Do what the function is supposed to do
console.log(`Processing event ${event} with age ${eventAge} ms.`);
// Retry failed function executions
constfailed=false;
if(failed){
callback('some error');
}else{
callback();
}
});

fromdatetimeimport datetime, timezone
# The 'python-dateutil' package must be included in requirements.txt.
fromdateutilimport parser
importfunctions_framework
@functions_framework.cloud_event
defavoid_infinite_retries(cloud_event):
"""Cloud Event Function that only executes within a certain
 time period after the triggering event.
 Args:
 cloud_event: The cloud event associated with the current trigger
 Returns:
 None; output is written to Stackdriver Logging
 """
 timestamp = cloud_event["time"]
 event_time = parser.parse(timestamp)
 event_age = (datetime.now(timezone.utc) - event_time).total_seconds()
 event_age_ms = event_age * 1000
 # Ignore events that are too old
 max_age_ms = 10000
 if event_age_ms > max_age_ms:
 print("Dropped {} (age {}ms)".format(cloud_event["id"], event_age_ms))
 return "Timeout"
 # Do what the function is supposed to do
 print("Processed {} (age {}ms)".format(cloud_event["id"], event_age_ms))
 return # To retry the execution, raise an exception here

// Package tips contains tips for writing Cloud Functions in Go.
packagetips
import(
"context"
"fmt"
"log"
"time"
"github.com/GoogleCloudPlatform/functions-framework-go/functions"
"github.com/cloudevents/sdk-go/v2/event"
)
funcinit(){
functions.CloudEvent("FiniteRetryPubSub",FiniteRetryPubSub)
}
// MessagePublishedData contains the full Pub/Sub message
// See the documentation for more details:
// https://cloud.google.com/eventarc/docs/cloudevents#pubsub
typeMessagePublishedDatastruct{
MessagePubSubMessage
}
// PubSubMessage is the payload of a Pub/Sub event.
// See the documentation for more details:
// https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage
typePubSubMessagestruct{
Data[]byte`json:"data"`
}
// FiniteRetryPubSub demonstrates how to avoid inifinite retries.
funcFiniteRetryPubSub(ctxcontext.Context,eevent.Event)error{
varmsgMessagePublishedData
iferr:=e.DataAs(&msg);err!=nil{
returnfmt.Errorf("event.DataAs: %w",err)
}
// Ignore events that are too old.
expiration:=e.Time().Add(10*time.Second)
iftime.Now().After(expiration){
log.Printf("event timeout: halting retries for expired event '%q'",e.ID())
returnnil
}
// Add your message processing logic.
returnprocessTheMessage(msg)
}

importcom.google.cloud.functions.CloudEventsFunction;
importio.cloudevents.CloudEvent;
importjava.time.Duration;
importjava.time.ZoneOffset;
importjava.time.ZonedDateTime;
importjava.util.logging.Logger;
publicclass RetryTimeoutimplementsCloudEventsFunction{
privatestaticfinalLoggerlogger=Logger.getLogger(RetryTimeout.class.getName());
privatestaticfinallongMAX_EVENT_AGE=10_000;
/**
 * Cloud Event Function that only executes within
 * a certain time period after the triggering event
 */
@Override
publicvoidaccept(CloudEventevent)throwsException{
ZonedDateTimeutcNow=ZonedDateTime.now(ZoneOffset.UTC);
ZonedDateTimetimestamp=event.getTime().atZoneSameInstant(ZoneOffset.UTC);
longeventAge=Duration.between(timestamp,utcNow).toMillis();
// Ignore events that are too old
if(eventAge > MAX_EVENT_AGE){
logger.info(String.format("Dropping event with timestamp %s.",timestamp));
return;
}
// Process events that are recent enough
// To retry this invocation, throw an exception here
logger.info(String.format("Processing event with timestamp %s.",timestamp));
}
}

usingCloudNative.CloudEvents;
usingGoogle.Cloud.Functions.Framework;
usingGoogle.Events.Protobuf.Cloud.PubSub.V1;
usingMicrosoft.Extensions.Logging;
usingSystem;
usingSystem.Threading;
usingSystem.Threading.Tasks;
namespaceTimeBoundedRetries;
publicclassFunction:ICloudEventFunction<MessagePublishedData>
{
privatestaticreadonlyTimeSpanMaxEventAge=TimeSpan.FromSeconds(10);
privatereadonlyILogger_logger;
// Note: for additional testability, use an injectable clock abstraction.
publicFunction(ILogger<Function>logger)=>
_logger=logger;
publicTaskHandleAsync(CloudEventcloudEvent,MessagePublishedDatadata,CancellationTokencancellationToken)
{
stringtextData=data.Message.TextData;
DateTimeOffsetutcNow=DateTimeOffset.UtcNow;
// Every PubSub CloudEvent will contain a timestamp.
DateTimeOffsettimestamp=cloudEvent.Time.Value;
DateTimeOffsetexpiry=timestamp+MaxEventAge;
// Ignore events that are too old.
if(utcNow > expiry)
{
_logger.LogInformation("Dropping PubSub message '{text}'",textData);
returnTask.CompletedTask;
}
// Process events that are recent enough.
// If this processing throws an exception, the message will be retried until either
// processing succeeds or the event becomes too old and is dropped by the code above.
_logger.LogInformation("Processing PubSub message '{text}'",textData);
returnTask.CompletedTask;
}
}

require"functions_framework"
FunctionsFramework.cloud_event"avoid_infinite_retries"do|event|
# Use the event timestamp to determine the event age.
event_age_secs=Time.now-event.time.to_time
event_age_ms=(event_age_secs*1000).to_i
max_age_ms=10_000
ifevent_age_ms > max_age_ms
# Ignore events that are too old.
logger.info"Dropped #{event.id} (age #{event_age_ms}ms)"
else
# Do what the function is supposed to do.
logger.info"Handling #{event.id} (age #{event_age_ms}ms)..."
failed=true
# Raise an exception to signal failure and trigger a retry.
raise"I failed!"iffailed
end
end

/**
 * This function shows an example method for avoiding infinite retries in
 * Google Cloud Functions. By default, functions configured to automatically
 * retry execution on failure will be retried indefinitely - causing an
 * infinite loop. To avoid this, we stop retrying executions (by not throwing
 * exceptions) for any events that are older than a predefined threshold.
 */
use Google\CloudFunctions\CloudEvent;
function avoidInfiniteRetries(CloudEvent $event): void
{
 $log = fopen(getenv('LOGGER_OUTPUT') ?: 'php://stderr', 'wb');
 $eventId = $event->getId();
 // The maximum age of events to process.
 $maxAge = 10; // 10 seconds
 // The age of the event being processed.
 $eventAge = time() - strtotime($event->getTime());
 // Ignore events that are too old
 if ($eventAge > $maxAge) {
 fwrite($log, 'Dropping event ' . $eventId . ' with age ' . $eventAge . ' seconds' . PHP_EOL);
 return;
 }
 // Do what the function is supposed to do
 fwrite($log, 'Processing event: ' . $eventId . ' with age ' . $eventAge . ' seconds' . PHP_EOL);
 // infinite_retries failed function executions
 $failed = true;
 if ($failed) {
 throw new Exception('Event ' . $eventId . ' failed; retrying...');
 }
}

constfunctions=require('@google-cloud/functions-framework');
/**
 * Register a Cloud Event Function that demonstrates
 * how to toggle retries using a promise
 *
 * @param {object} event The Cloud Event for the function trigger.
 */
functions.cloudEvent('retryPromise',cloudEvent=>{
// The Pub/Sub event payload is passed as the CloudEvent's data payload.
// See the documentation for more details:
// https://cloud.google.com/eventarc/docs/cloudevents#pubsub
constbase64PubsubMessage=cloudEvent.data.message.data;
constjsonString=Buffer.from(base64PubsubMessage,'base64').toString();
consttryAgain=JSON.parse(jsonString).retry;
if(tryAgain){
thrownewError('Retrying...');
}else{
console.error('Not retrying...');
returnPromise.resolve();
}
});
/**
 * Cloud Event Function that demonstrates
 * how to toggle retries using a callback
 *
 * @param {object} event The Cloud Event for the function trigger.
 * @param {function} callback The callback function.
 */
functions.cloudEvent('retryCallback',(cloudEvent,callback)=>{
// The Pub/Sub event payload is passed as the CloudEvent's data payload.
// See the documentation for more details:
// https://cloud.google.com/eventarc/docs/cloudevents#pubsub
constbase64PubsubMessage=cloudEvent.data.message.data;
constjsonString=Buffer.from(base64PubsubMessage,'base64').toString();
consttryAgain=JSON.parse(jsonString).retry;
consterr=newError('Error!');
if(tryAgain){
console.error('Retrying:',err);
callback(err);
}else{
console.error('Not retrying:',err);
callback();
}
});

importbase64
importjson
importfunctions_framework
fromgoogle.cloudimport error_reporting
error_client = error_reporting.Client ()
@functions_framework.cloud_event
defretry_or_not(cloud_event):
"""Cloud Event Function that demonstrates how to toggle retries.
 Args:
 cloud_event: The cloud event with a Pub/Sub data payload
 Returns:
 None; output is written to Stackdriver Logging
 """
 # The Pub/Sub event payload is passed as the CloudEvent's data payload.
 # See the documentation for more details:
 # https://cloud.google.com/eventarc/docs/cloudevents#pubsub
 encoded_pubsub_message = cloud_event.data["message"]["data"]
 # Retry based on a user-defined parameter
 try_again = json.loads(base64.b64decode(encoded_pubsub_message).decode())["retry"]
 try:
 raise RuntimeError("I failed you")
 except RuntimeError:
 error_client.report_exception ()
 if try_again:
 raise # Raise the exception and try again
 else:
 pass # Swallow the exception and don't retry

packagetips
import(
"context"
"errors"
"fmt"
"log"
"github.com/GoogleCloudPlatform/functions-framework-go/functions"
"github.com/cloudevents/sdk-go/v2/event"
)
funcinit(){
functions.CloudEvent("RetryPubSub",RetryPubSub)
}
// MessagePublishedData contains the full Pub/Sub message
// See the documentation for more details:
// https://cloud.google.com/eventarc/docs/cloudevents#pubsub
typeMessagePublishedDatastruct{
MessagePubSubMessage
}
// PubSubMessage is the payload of a Pub/Sub event.
// See the documentation for more details:
// https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage
typePubSubMessagestruct{
Data[]byte`json:"data"`
}
// RetryPubSub demonstrates how to toggle using retries.
funcRetryPubSub(ctxcontext.Context,eevent.Event)error{
varmsgMessagePublishedData
iferr:=e.DataAs(&msg);err!=nil{
returnfmt.Errorf("event.DataAs: %w",err)
}
name:=string(msg.Message.Data)
ifname==""{
name="World"
}
// A misconfigured client will stay broken until the function is redeployed.
client,err:=MisconfiguredDataClient()
iferr!=nil{
log.Printf("MisconfiguredDataClient (retry denied): %v",err)
// A nil return indicates that the function does not need a retry.
returnnil
}
// Runtime error might be resolved with a new attempt.
iferr=FailedWriteOperation(client,name);err!=nil{
log.Printf("FailedWriteOperation (retry expected): %v",err)
// A non-nil return indicates that a retry is needed.
returnerr
}
returnnil
}

importcom.google.cloud.functions.CloudEventsFunction;
importcom.google.gson.Gson;
importcom.google.gson.JsonElement;
importcom.google.gson.JsonObject;
importfunctions.eventpojos.PubSubBody;
importio.cloudevents.CloudEvent;
importjava.nio.charset.StandardCharsets;
importjava.util.Base64;
importjava.util.logging.Logger;
publicclass RetryPubSubimplementsCloudEventsFunction{
privatestaticfinalLoggerlogger=Logger.getLogger(RetryPubSub.class.getName());
// Use Gson (https://github.com/google/gson) to parse JSON content.
privatestaticfinalGsongson=newGson();
@Override
publicvoidaccept(CloudEventevent)throwsException{
if(event.getData()==null){
logger.warning("No data found in event!");
return;
}
// Extract Cloud Event data and convert to PubSubBody
StringcloudEventData=newString(event.getData().toBytes(),StandardCharsets.UTF_8);
PubSubBodybody=gson.fromJson(cloudEventData,PubSubBody.class);
StringencodedData=body.getMessage().getData();
StringdecodedData=
newString(Base64.getDecoder().decode(encodedData),StandardCharsets.UTF_8);
// Retrieve and decode PubSubMessage data into a JsonElement.
// Function is expecting a user-supplied JSON message which determines whether
// to retry or not.
JsonElementjsonPubSubMessageElement=gson.fromJson(decodedData,JsonElement.class);
booleanretry=false;
// Get the value of the "retry" JSON parameter, if one exists
if(jsonPubSubMessageElement!=null && jsonPubSubMessageElement.isJsonObject()){
JsonObjectjsonPubSubMessageObject=jsonPubSubMessageElement.getAsJsonObject();
if(jsonPubSubMessageObject.has("retry")
 && jsonPubSubMessageObject.get("retry").getAsBoolean()){
retry=true;
}
}
// Retry if appropriate
if(retry){
// Throwing an exception causes the execution to be retried
thrownewRuntimeException("Retrying...");
}else{
logger.info("Not retrying...");
}
}
}

usingCloudNative.CloudEvents;
usingGoogle.Cloud.Functions.Framework;
usingGoogle.Events.Protobuf.Cloud.PubSub.V1;
usingMicrosoft.Extensions.Logging;
usingSystem;
usingSystem.Text.Json;
usingSystem.Threading;
usingSystem.Threading.Tasks;
namespaceRetry;
publicclassFunction:ICloudEventFunction<MessagePublishedData>
{
privatereadonlyILogger_logger;
publicFunction(ILogger<Function>logger)=>
_logger=logger;
publicTaskHandleAsync(CloudEventcloudEvent,MessagePublishedDatadata,CancellationTokencancellationToken)
{
boolretry=false;
stringtext=data.Message?.TextData;
// Get the value of the "retry" JSON parameter, if one exists.
if(!string.IsNullOrEmpty(text))
{
JsonElementelement=JsonSerializer.Deserialize<JsonElement>(data.Message.TextData);
retry=element.TryGetProperty("retry",outvarproperty)&&
property.ValueKind==JsonValueKind.True;
}
// Throwing an exception causes the execution to be retried.
if(retry)
{
thrownewInvalidOperationException("Retrying...");
}
else
{
_logger.LogInformation("Not retrying...");
}
returnTask.CompletedTask;
}
}

require"functions_framework"
FunctionsFramework.cloud_event"retry_or_not"do|event|
try_again=event.data["retry"]
begin
# Simulate a failure
raise"I failed!"
rescueRuntimeError=>e
logger.warn"Caught an error: #{e}"
iftry_again
# Raise an exception to return a 500 and trigger a retry.
logger.info"Trying again..."
raiseex
else
# Return normally to end processing of this event.
logger.info"Giving up."
end
end
end

use Google\CloudFunctions\CloudEvent;
function tipsRetry(CloudEvent $event): void
{
 $cloudEventData = $event->getData();
 $pubSubData = $cloudEventData['message']['data'];
 $json = json_decode(base64_decode($pubSubData), true);
 // Determine whether to retry the invocation based on a parameter
 $tryAgain = $json['some_parameter'];
 if ($tryAgain) {
 /**
 * Functions with automatic retries enabled should throw exceptions to
 * indicate intermittent failures that a retry might fix. In this
 * case, a thrown exception will cause the original function
 * invocation to be re-sent.
 */
 throw new Exception('Intermittent failure occurred; retrying...');
 }
 /**
 * If a function with retries enabled encounters a non-retriable
 * failure, it should return *without* throwing an exception.
 */
 $log = fopen(getenv('LOGGER_OUTPUT') ?: 'php://stderr', 'wb');
 fwrite($log, 'Not retrying' . PHP_EOL);
}