Skip to content

HTTP-122 Retry for source lookup table #148

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

Merged
merged 22 commits into from
May 6, 2025
Merged

HTTP-122 Retry for source lookup table #148

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
610d4b9
retry for source lookup table
maciejmaciejko-gid Mar 26, 2025
16ffc35
changelog, readme, refactor
maciejmaciejko-gid Mar 26, 2025
a547bde
refactor after review
maciejmaciejko-gid Mar 27, 2025
e0c7bcc
increasing code coverage
maciejmaciejko-gid Mar 27, 2025
b67567a
Merge branch 'main' into feature/retries
maciejmaciejko-gid Mar 27, 2025
680723b
increasing code coverage
maciejmaciejko-gid Mar 27, 2025
9485f27
exposing retry metrics
maciejmaciejko-gid Mar 28, 2025
853d23d
logging optimisation
maciejmaciejko-gid Mar 28, 2025
0c5097b
removed unused metrics
maciejmaciejko-gid Mar 28, 2025
75eaa2d
handling of max-retries param, readme fix, metric names convention
maciejmaciejko-gid Mar 28, 2025
65de8b8
import order and style check
maciejmaciejko-gid Mar 31, 2025
9753dea
clean up
maciejmaciejko-gid Mar 31, 2025
c3ed544
not managed retries
maciejmaciejko-gid Mar 31, 2025
0a56ba0
reformat
maciejmaciejko-gid Mar 31, 2025
3cf56da
reformat
maciejmaciejko-gid Mar 31, 2025
b08575d
reformat
maciejmaciejko-gid Mar 31, 2025
e497587
changelog fix
maciejmaciejko-gid Mar 31, 2025
dfb1133
clean up
maciejmaciejko-gid Apr 15, 2025
d9c4257
fail on error
maciejmaciejko-gid Apr 16, 2025
195f4c3
ignored response codes can be omitted in success codes
maciejmaciejko-gid May 6, 2025
95aa14e
ignored response codes can be omitted in success codes
maciejmaciejko-gid May 6, 2025
72dc8ab
refactor
maciejmaciejko-gid May 6, 2025
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
7 changes: 7 additions & 0 deletions src/main/java/com/getindata/connectors/http/internal/PollingClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import java.util.Collection;

import org.apache.flink.table.data.RowData;
import org.apache.flink.table.functions.FunctionContext;

/**
* A client that is used to get enrichment data from external component.
Expand All @@ -15,4 +16,10 @@ public interface PollingClient<T> {
* @return an optional result of data lookup.
*/
Collection<T> pull(RowData lookupRow);

/**
* Initialize the client.
* @param ctx function context
*/
void open(FunctionContext ctx);
}
20 changes: 17 additions & 3 deletions src/main/java/com/getindata/connectors/http/internal/retry/HttpClientWithRetry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,12 @@
import com.getindata.connectors.http.internal.status.HttpResponseChecker;
import io.github.resilience4j.retry.Retry;
import io.github.resilience4j.retry.RetryConfig;
import io.github.resilience4j.retry.RetryRegistry;
import lombok.Builder;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.metrics.MetricGroup;

import java.io.IOException;
import java.net.http.HttpClient;
Expand All @@ -19,26 +21,38 @@
public class HttpClientWithRetry {

private final HttpClient httpClient;
private final RetryConfig retryConfig;
@Getter
private final HttpResponseChecker responseChecker;
private final Retry retry;

@Builder
HttpClientWithRetry(HttpClient httpClient,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am wondering what happens with OIDC, the short lived bearer token may need to be regenerated if the retries occur after the token has expired). Is this regeneration check done for the retries?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The request is created only once, but OIDC processor (responsible for setting bearer token in request) is called on every retry.

RetryConfig retryConfig,
HttpResponseChecker responseChecker) {
this.httpClient = httpClient;
this.responseChecker = responseChecker;
this.retryConfig = RetryConfig.from(retryConfig)
retryConfig = RetryConfig.from(retryConfig)
.retryExceptions(IOException.class, RetryHttpRequestException.class)
.build();
this.retry = RetryRegistry.ofDefaults().retry("http-lookup-connector", retryConfig);
}

public void registerMetrics(MetricGroup metrics){
var group = metrics.addGroup("http_lookup_connector");
group.gauge("failed_calls_with_retry_attempt_count",
() -> retry.getMetrics().getNumberOfFailedCallsWithRetryAttempt());
group.gauge("failed_calls_without_retry_attempt_count",
() -> retry.getMetrics().getNumberOfFailedCallsWithoutRetryAttempt());
group.gauge("success_calls_with_retry_attempt",
() -> retry.getMetrics().getNumberOfSuccessfulCallsWithRetryAttempt());
group.gauge("success_calls_without_retry_attempt",
() -> retry.getMetrics().getNumberOfSuccessfulCallsWithoutRetryAttempt());
}

public <T> HttpResponse<T> send(
Supplier<HttpRequest> requestSupplier,
HttpResponse.BodyHandler<T> responseBodyHandler
) throws IOException, InterruptedException, HttpStatusCodeValidationFailedException {
var retry = Retry.of("http-lookup-connector", retryConfig);
try {
try {
return Retry.decorateCheckedSupplier(retry, () -> task(requestSupplier, responseBodyHandler)).apply();
Expand Down
13 changes: 7 additions & 6 deletions ...ain/java/com/getindata/connectors/http/internal/table/lookup/HttpTableLookupFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
package com.getindata.connectors.http.internal.table.lookup;

import java.util.Collection;
import java.util.concurrent.atomic.AtomicInteger;

import com.getindata.connectors.http.internal.PollingClient;
import com.getindata.connectors.http.internal.PollingClientFactory;
import com.getindata.connectors.http.internal.utils.SerializationSchemaUtils;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
Expand All @@ -12,9 +12,8 @@
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.LookupFunction;

import com.getindata.connectors.http.internal.PollingClient;
import com.getindata.connectors.http.internal.PollingClientFactory;
import com.getindata.connectors.http.internal.utils.SerializationSchemaUtils;
import java.util.Collection;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class HttpTableLookupFunction extends LookupFunction {
Expand Down Expand Up @@ -59,6 +58,8 @@ public void open(FunctionContext context) throws Exception {

context.getMetricGroup()
.gauge("http-table-lookup-call-counter", () -> localHttpCallCounter.intValue());

client.open(context);
}

@Override
Expand Down
6 changes: 6 additions & 0 deletions ...in/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.util.ConfigurationException;
import org.apache.flink.util.StringUtils;

Expand Down Expand Up @@ -84,6 +85,11 @@ public JavaNetHttpPollingClient(
validateIgnoredResponseCodes(this.httpClient.getResponseChecker());
}

public void open(FunctionContext context) {
httpClient.registerMetrics(context.getMetricGroup());
}


@Override
public Collection<RowData> pull(RowData lookupRow) {
try {
Expand Down
Loading