gCubeSystem
/
event-publisher-library
15
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Better handling of exceptions and retrying behavior in case of read timeout. (Default connection timeout is 10s and read timeout now is 60s)

This commit is contained in:
Mauro Mugnaini 2024-04-22 16:17:59 +02:00
parent d761dac8ac
commit 7790eb127c
Signed by: mauro.mugnaini
GPG Key ID: 2440CFD0EB321EA8
6 changed files with 75 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.classpath
View File

@ -15,7 +15,6 @@
</classpathentry>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-11">
<attributes>
<attribute name="module" value="true"/>
<attribute name="maven.pomderived" value="true"/>
</attributes>
</classpathentry>
@ -32,5 +31,11 @@
<attribute name="maven.pomderived" value="true"/>
</attributes>
</classpathentry>
<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources">
<attributes>
<attribute name="maven.pomderived" value="true"/>
<attribute name="optional" value="true"/>
</attributes>
</classpathentry>
<classpathentry kind="output" path="target/classes"/>
</classpath>

3
.settings/org.eclipse.wst.common.project.facet.core.xml
View File

@ -1,6 +1,5 @@
<?xml version="1.0" encoding="UTF-8"?>
<faceted-project>
<runtime name="Liferay v6.2 (Tomcat 7)"/>
<installed facet="jst.utility" version="1.0"/>
<installed facet="java" version="1.8"/>
<installed facet="java" version="11"/>
</faceted-project>

1
CHANGELOG.md
View File

@ -6,6 +6,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
- Restored correct behavior for event publishing with workflow id only sent back
- Extracted `AbstractHTTPWithJWTTokenAuthEventSender` class for easy subclassing
- Added new outcome check methods to inspect last send and last check actions results and some other facilities
- Better handling of exceptions and retrying behavior in case of read timeout. (Default connection timeout is 10s and read timeout now is 60s)
## [v1.1.0]
- Added `BufferedEventProcessor` to manual send bunch of events and controlling their output status (#23628)

7
pom.xml
View File

@ -7,7 +7,7 @@
<parent>
<artifactId>maven-parent</artifactId>
<groupId>org.gcube.tools</groupId>
<version>1.2.1-SNAPSHOT</version>
<version>1.2.0</version>
<relativePath />
</parent>
@ -33,6 +33,11 @@
<developerConnection>scm:git:https://code-repo.d4science.org/gCubeSystem/${project.artifactId}.git</developerConnection>
<url>https://code-repo.d4science.org/gCubeSystem/${project.artifactId}</url>
</scm>
<properties>
<java.version>11</java.version>
<maven.compiler.source>${java.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
</properties>
<build>
<plugins>

4
src/main/java/org/gcube/event/publisher/AbstractEventPublisher.java
View File

@ -44,7 +44,7 @@ public abstract class AbstractEventPublisher implements EventPublisher {
@Override
public int getLastPublishEventHTTPResponseCode() {
return eventSender.getLastSendHTTPResponseCode();
return getEventSender().getLastSendHTTPResponseCode();
}
@Override
@ -88,7 +88,7 @@ public abstract class AbstractEventPublisher implements EventPublisher {
@Override
public int getLastCheckHTTPResponseCode() {
return eventSender.getLastRetrieveHTTPResponseCode();
return getEventSender().getLastRetrieveHTTPResponseCode();
}
protected abstract EventSender createEventSender();

100
src/main/java/org/gcube/event/publisher/AbstractHTTPWithJWTTokenAuthEventSender.java
View File

@ -6,6 +6,7 @@ import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.SocketTimeoutException;
import java.net.URL;
import java.util.Arrays;
@ -32,21 +33,20 @@ public abstract class AbstractHTTPWithJWTTokenAuthEventSender implements EventSe
public AbstractHTTPWithJWTTokenAuthEventSender(URL baseEndpointURL, String clientId, String clientSecret,
URL tokenURL) {
this(baseEndpointURL, clientId, clientSecret, tokenURL, HTTPVerb.DEFAULT_CONNECTION_TIMEOUT,
HTTPVerb.DEFAULT_READ_TIMEOUT);
}
public AbstractHTTPWithJWTTokenAuthEventSender(URL baseEndpointURL, String clientId, String clientSecret,
URL tokenURL, int connectionTimeout, int readTimeout) {
super();
this.baseEndpointURL = baseEndpointURL;
this.clientId = clientId;
this.clientSecret = clientSecret;
this.tokenURL = tokenURL;
this.connectionTimeout = getDefaultConnectionTimeout();
this.readTimeout = getDefaultReadTimeout();
}
protected int getDefaultReadTimeout() {
return HTTPVerb.DEFAULT_READ_TIMEOUT;
}
protected int getDefaultConnectionTimeout() {
return HTTPVerb.DEFAULT_CONNECTION_TIMEOUT;
this.connectionTimeout = connectionTimeout;
this.readTimeout = readTimeout;
}
public int getReadTimeout() {
@ -112,7 +112,7 @@ public abstract class AbstractHTTPWithJWTTokenAuthEventSender implements EventSe
public abstract class HTTPVerb {
protected static final int DEFAULT_CONNECTION_TIMEOUT = 10000;
protected static final int DEFAULT_READ_TIMEOUT = 5000;
protected static final int DEFAULT_READ_TIMEOUT = 60000;
protected URL baseEndpoint;
protected int httpResponseCode = -1;
@ -130,8 +130,10 @@ public abstract class AbstractHTTPWithJWTTokenAuthEventSender implements EventSe
public class HTTPPost extends HTTPVerb implements Runnable {
private static final int PAUSE_INCREMENT_FACTOR = 2;
private static final long MAX_RETRYINGS = 2;
private static final int FIRST_PAUSE = 7500; // ms
private static final int PAUSE_INCREMENT_FACTOR = 4;
private static final long MAX_RETRYINGS = 5;
private static final boolean RETRY_ON_READ_TIMEOUT = false;
private final int[] RETRY_CODES = { HttpURLConnection.HTTP_BAD_GATEWAY,
HttpURLConnection.HTTP_CLIENT_TIMEOUT, HttpURLConnection.HTTP_GATEWAY_TIMEOUT,
@ -139,7 +141,7 @@ public abstract class AbstractHTTPWithJWTTokenAuthEventSender implements EventSe
private Event event;
private String result;
private long actualPause = 1;
private long actualPause = FIRST_PAUSE;
private long retryings = 0;
public HTTPPost(URL baseEndpoint, Event event) {
@ -154,7 +156,8 @@ public abstract class AbstractHTTPWithJWTTokenAuthEventSender implements EventSe
if (baseEndpoint.toString().endsWith("/")) {
try {
log.debug("Removing trailing slash from base URL since the endpoint computation API changed");
eventEndpoint = new URL(baseEndpoint.toString().substring(0, baseEndpoint.toString().length() - 1));
eventEndpoint = new URL(
baseEndpoint.toString().substring(0, baseEndpoint.toString().length() - 1));
} catch (MalformedURLException e) {
log.error("Cannot remove trailing slash from base endpoint URL", e);
return;
@ -206,45 +209,60 @@ public abstract class AbstractHTTPWithJWTTokenAuthEventSender implements EventSe
} else {
isr = new InputStreamReader(connection.getErrorStream(), "UTF-8");
}
BufferedReader br = new BufferedReader(isr);
String line = null;
while ((line = br.readLine()) != null) {
sb.append("\n" + line);
try {
BufferedReader br = new BufferedReader(isr);
String line = null;
while ((line = br.readLine()) != null) {
sb.append("\n" + line);
}
br.close();
isr.close();
sb.deleteCharAt(0);
result = sb.toString();
} catch (Exception e) {
log.error("Error reading response", e);
break;
}
br.close();
isr.close();
sb.deleteCharAt(0);
result = sb.toString();
if (OK) {
log.debug("[{}] Event publish for {} is OK", httpResponseCode, event.getName());
} else {
log.trace("Response message from server:\n{}", result);
if (shouldRetryWithCode(httpResponseCode)) {
if (retryings <= MAX_RETRYINGS) {
log.warn("[{}] Event publish ERROR, retrying in {} seconds", httpResponseCode,
actualPause);
Thread.sleep(actualPause * 1000);
log.debug("Start retrying event publish: {}", event.getName());
actualPause *= PAUSE_INCREMENT_FACTOR;
retryings += 1;
} else {
log.error("[{}] Event publish ERROR, exhausted tries after {} retryings",
httpResponseCode,
retryings);
break;
}
log.warn("[{}] Event publish ERROR", httpResponseCode);
} else {
log.info("[{}] Event publish ERROR but should not retry with this HTTP code",
log.info("[{}] Event publish ERROR but not retrying to send it with this HTTP code",
httpResponseCode);
break;
}
}
} catch (SocketTimeoutException e) {
log.error("Read timeout POSTing JSON to: " + eventEndpoint, e);
if (RETRY_ON_READ_TIMEOUT) {
log.trace("Event will be re-sent accodingly to retryes");
} else {
log.trace("Event will be not re-sent");
break;
}
} catch (IOException | OpenIdConnectRESTHelperException e) {
log.error("POSTing JSON to: " + eventEndpoint, e);
break;
log.error("I/O Error POSTing JSON to: " + eventEndpoint, e);
}
if (!OK) {
if (retryings <= MAX_RETRYINGS) {
log.info("Retrying to send event in {} ms", actualPause);
Thread.sleep(actualPause);
log.debug("Start retrying event publish: {}", event.getName());
actualPause *= PAUSE_INCREMENT_FACTOR;
retryings += 1;
} else {
log.error("[{}] Event publish ERROR, exhausted tries after {} retryings",
httpResponseCode,
retryings);
break;
}
}
} while (!OK);
} catch (InterruptedException e) {