"rate-limit" the number of events we process per timeframe

Author: marko-bekhta

src/main/java/org/hibernate/infra/sync/jira/JiraConfig.java

@@ -34,6 +34,15 @@ interface JiraProjectGroup {
 
 		Scheduled scheduled();
 
+		EventProcessing processing();
+	}
+
+	interface EventProcessing {
+		@WithDefault("5")
+		int eventsPerTimeframe();
+
+		@WithDefault("1")
+		int timeframeInSeconds();
 	}
 
 	interface Scheduled {

src/main/java/org/hibernate/infra/sync/jira/ProcessingConfig.java

@@ -5,9 +5,9 @@
 
 @ConfigMapping(prefix = "processing.events")
 public interface ProcessingConfig {
-	@WithDefault("1000")
+	@WithDefault("10000")
 	int queueSize();
 
-	@WithDefault("20")
+	@WithDefault("5")
 	int threads();
 }

src/main/java/org/hibernate/infra/sync/jira/service/jira/HandlerProjectContext.java

@@ -14,7 +14,7 @@
 
 import io.quarkus.logging.Log;
 
-public final class HandlerProjectContext {
+public final class HandlerProjectContext implements AutoCloseable {
 
 	// JIRA REST API creates upto 50 issues at a time:
 	// https://developer.atlassian.com/cloud/jira/platform/rest/v2/api-group-issues/#api-rest-api-2-issue-bulk-post
@@ -26,18 +26,18 @@ public final class HandlerProjectContext {
 	private final String projectGroupName;
 	private final JiraRestClient sourceJiraClient;
 	private final JiraRestClient destinationJiraClient;
-	private final JiraConfig.JiraProjectGroup projectGroup;
+	private final HandlerProjectGroupContext projectGroupContext;
 	private final JiraConfig.JiraProject project;
 	private final AtomicLong currentIssueKeyNumber;
 	private final JiraIssueBulk bulk;
 
 	public HandlerProjectContext(String projectName, String projectGroupName, JiraRestClient sourceJiraClient,
-			JiraRestClient destinationJiraClient, JiraConfig.JiraProjectGroup projectGroup) {
+			JiraRestClient destinationJiraClient, HandlerProjectGroupContext projectGroupContext) {
 		this.projectName = projectName;
 		this.projectGroupName = projectGroupName;
 		this.sourceJiraClient = sourceJiraClient;
 		this.destinationJiraClient = destinationJiraClient;
-		this.projectGroup = projectGroup;
+		this.projectGroupContext = projectGroupContext;
 		this.project = projectGroup().projects().get(projectName());
 		this.currentIssueKeyNumber = new AtomicLong(getCurrentLatestJiraIssueKeyNumber());
 		this.bulk = new JiraIssueBulk(createIssuePlaceholder(), ISSUES_PER_REQUEST);
@@ -64,7 +64,7 @@ public JiraRestClient destinationJiraClient() {
 	}
 
 	public JiraConfig.JiraProjectGroup projectGroup() {
-		return projectGroup;
+		return projectGroupContext.projectGroup();
 	}
 
 	public AtomicLong currentIssueKeyNumber() {
@@ -129,6 +129,10 @@ public void createNextPlaceholderBatch(Long upToKeyNumber) {
 		}
 	}
 
+	public void startProcessingEvent() throws InterruptedException {
+		projectGroupContext.startProcessingEvent();
+	}
+
 	private boolean requiredIssueKeyNumberShouldBeAvailable(Long key) {
 		return currentIssueKeyNumber.get() >= key;
 	}
@@ -156,8 +160,12 @@ private JiraIssue createIssuePlaceholder() {
 	public String toString() {
 		return "HandlerProjectContext[" + "projectName=" + projectName + ", " + "projectGroupName=" + projectGroupName
 				+ ", " + "sourceJiraClient=" + sourceJiraClient + ", " + "destinationJiraClient="
-				+ destinationJiraClient + ", " + "projectGroup=" + projectGroup + ", " + "currentIssueKeyNumber="
-				+ currentIssueKeyNumber + ']';
+				+ destinationJiraClient + ", " + "projectGroup=" + projectGroupContext.projectGroup() + ", "
+				+ "currentIssueKeyNumber=" + currentIssueKeyNumber + ']';
 	}
 
+	@Override
+	public void close() {
+		projectGroupContext.close();
+	}
 }

src/main/java/org/hibernate/infra/sync/jira/service/jira/HandlerProjectGroupContext.java

@@ -0,0 +1,43 @@
+package org.hibernate.infra.sync.jira.service.jira;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import org.hibernate.infra.sync.jira.JiraConfig;
+
+public final class HandlerProjectGroupContext implements AutoCloseable {
+
+	private final ScheduledExecutorService rateLimiterExecutor = Executors.newScheduledThreadPool(1);
+	private final Semaphore rateLimiter;
+	private final JiraConfig.JiraProjectGroup projectGroup;
+
+	public HandlerProjectGroupContext(JiraConfig.JiraProjectGroup projectGroup) {
+		this.projectGroup = projectGroup;
+
+		this.rateLimiter = new Semaphore(projectGroup.processing().eventsPerTimeframe());
+		rateLimiterExecutor.scheduleAtFixedRate(() -> {
+			rateLimiter.drainPermits();
+			rateLimiter.release(5);
+		}, projectGroup.processing().timeframeInSeconds(), projectGroup.processing().timeframeInSeconds(),
+				TimeUnit.SECONDS);
+	}
+
+	public void startProcessingEvent() throws InterruptedException {
+		rateLimiter.acquire(1);
+	}
+
+	public JiraConfig.JiraProjectGroup projectGroup() {
+		return projectGroup;
+	}
+
+	@Override
+	public void close() {
+		// when requesting to close the context we aren't expecting to process any other
+		// events hence there's no point in continuing "releasing" more "permits":
+		if (!rateLimiterExecutor.isShutdown()) {
+			rateLimiterExecutor.shutdownNow();
+		}
+	}
+}

src/main/java/org/hibernate/infra/sync/jira/service/jira/JiraService.java

@@ -34,6 +34,7 @@
 import io.quarkus.scheduler.Scheduler;
 import io.quarkus.vertx.http.ManagementInterface;
 import io.vertx.core.json.JsonObject;
+import jakarta.annotation.PreDestroy;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.enterprise.event.Observes;
 import jakarta.inject.Inject;
@@ -59,10 +60,10 @@ public JiraService(ProcessingConfig processingConfig, JiraConfig jiraConfig, Rep
 		for (var entry : jiraConfig.projectGroup().entrySet()) {
 			JiraRestClient source = JiraRestClientBuilder.of(entry.getValue().source());
 			JiraRestClient destination = JiraRestClientBuilder.of(entry.getValue().destination());
-
+			HandlerProjectGroupContext groupContext = new HandlerProjectGroupContext(entry.getValue());
 			for (var project : entry.getValue().projects().entrySet()) {
-				contextMap.put(project.getKey(), new HandlerProjectContext(project.getKey(), entry.getKey(), source,
-						destination, entry.getValue()));
+				contextMap.put(project.getKey(),
+						new HandlerProjectContext(project.getKey(), entry.getKey(), source, destination, groupContext));
 			}
 		}
 
@@ -217,6 +218,25 @@ public void syncLastUpdated(String projectGroup) {
 		}
 	}
 
+	@PreDestroy
+	public void finishProcessingAndShutdown() {
+		try {
+			executor.shutdown();
+			if (!executor.awaitTermination(2, TimeUnit.MINUTES)) {
+				Log.infof("Not all events were processed before the shutdown");
+			}
+			for (HandlerProjectContext context : contextPerProject.values()) {
+				try {
+					context.close();
+				} catch (Exception e) {
+					Log.errorf(e, "Error closing context %s: %s", context, e.getMessage());
+				}
+			}
+		} catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		}
+	}
+
 	private void syncByQuery(String query, HandlerProjectContext context) {
 		JiraIssues issues = null;
 		int start = 0;

src/main/java/org/hibernate/infra/sync/jira/service/jira/client/JiraRestClientBuilder.java

@@ -24,8 +24,7 @@ public static JiraRestClient of(JiraConfig.Instance jira) {
 		Map<String, String> headers = jira.loginKind().headers(jiraUser.email(), jiraUser.token());
 
 		QuarkusRestClientBuilder builder = QuarkusRestClientBuilder.newBuilder().baseUri(jira.apiUri())
-				// specifying a timeout of 0 represents infinity
-				.connectTimeout(0, TimeUnit.HOURS).readTimeout(0, TimeUnit.HOURS)
+				.connectTimeout(5, TimeUnit.MINUTES).readTimeout(10, TimeUnit.MINUTES)
 				.clientHeadersFactory((incomingHeaders, clientOutgoingHeaders) -> {
 					for (var entry : headers.entrySet()) {
 						clientOutgoingHeaders.add(entry.getKey(), entry.getValue());

src/main/java/org/hibernate/infra/sync/jira/service/jira/handler/JiraEventHandler.java

@@ -148,9 +148,13 @@ private static Map<String, String> createMapping(List<JiraSimpleObject> source,
 	@Override
 	public final void run() {
 		try {
+			context.startProcessingEvent();
 			doRun();
 		} catch (RuntimeException e) {
 			failureCollector.critical("Failed to handled the event", e);
+		} catch (InterruptedException e) {
+			failureCollector.critical("Interrupted while waiting in the queue", e);
+			Thread.currentThread().interrupt();
 		} finally {
 			failureCollector.close();
 		}

src/test/java/org/hibernate/infra/sync/jira/handler/IssueTest.java

@@ -7,6 +7,7 @@
 import org.hibernate.infra.sync.jira.JiraConfig;
 import org.hibernate.infra.sync.jira.mock.SampleJiraRestClient;
 import org.hibernate.infra.sync.jira.service.jira.HandlerProjectContext;
+import org.hibernate.infra.sync.jira.service.jira.HandlerProjectGroupContext;
 import org.hibernate.infra.sync.jira.service.jira.handler.JiraCommentDeleteEventHandler;
 import org.hibernate.infra.sync.jira.service.jira.handler.JiraCommentUpsertEventHandler;
 import org.hibernate.infra.sync.jira.service.jira.handler.JiraIssueDeleteEventHandler;
@@ -15,6 +16,7 @@
 import org.hibernate.infra.sync.jira.service.jira.handler.JiraIssueUpsertEventHandler;
 import org.hibernate.infra.sync.jira.service.reporting.ReportingConfig;
 
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -44,7 +46,12 @@ class IssueTest {
 	@BeforeEach
 	void setUp() {
 		context = new HandlerProjectContext("JIRATEST1", PROJECT_GROUP_NAME, source, destination,
-				jiraConfig.projectGroup().get(PROJECT_GROUP_NAME));
+				new HandlerProjectGroupContext(jiraConfig.projectGroup().get(PROJECT_GROUP_NAME)));
+	}
+
+	@AfterEach
+	void tearDown() {
+		context.close();
 	}
 
 	@Test