Commit 6dfe559c authored by Rustam Lotsmanenko (EPAM)'s avatar Rustam Lotsmanenko (EPAM)
Browse files

change subscriber admin client closing from try with resources to spring shutdown hook

parent 7a8e4a46
Pipeline #72773 failed with stages
in 27 minutes and 42 seconds
......@@ -23,33 +23,52 @@ import com.google.pubsub.v1.ExpirationPolicy;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.PushConfig;
import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.opengroup.osdu.core.common.model.tenant.TenantInfo;
import org.opengroup.osdu.register.utils.AppServiceConfig;
import org.opengroup.osdu.register.subscriber.model.Subscription;
import org.opengroup.osdu.register.utils.AppServiceConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Slf4j
@Component
public class PushSubscription {
private static final int ACK_DEADLINE_SECONDS = 60;
private static final int MESSAGE_RETENTION_SECONDS = 432000;
private static final int SUBSCRIPTION_EXPIRATION_SECONDS = 31540000;
private static final int AWAIT_TERMINATION_DURATION = 20;
@Autowired
private AppServiceConfig serviceConfig;
private SubscriptionAdminClient subscriptionAdminClient;
@PostConstruct
public void setUpSubscriptionAdminClient() throws IOException {
try {
this.subscriptionAdminClient = SubscriptionAdminClient.create();
} catch (IOException e) {
log.error("Exception occurred while creating subscription admin client", e);
throw e;
}
}
public Subscription create(Subscription subscription, TenantInfo tenant,
String masterGcpId, String serviceIdentityEmail)
throws IOException {
String masterGcpId, String serviceIdentityEmail)
throws IOException {
createPushSubscription(
tenant.getProjectId(),
masterGcpId,
subscription.getTopic(),
subscription.getNotificationId(),
serviceConfig.getRecordsChangePubsubEndpoint(),
serviceIdentityEmail);
tenant.getProjectId(),
masterGcpId,
subscription.getTopic(),
subscription.getNotificationId(),
serviceConfig.getRecordsChangePubsubEndpoint(),
serviceIdentityEmail);
return subscription;
}
......@@ -65,25 +84,33 @@ public class PushSubscription {
}
private com.google.pubsub.v1.Subscription createPushSubscription(String tenantProjectId, String servicesProjectId, String topicId,
String subscriptionId, String url, String serviceIdentityEmail) throws IOException {
try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {
ProjectTopicName topicName = ProjectTopicName.of(tenantProjectId, topicId);
ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of(servicesProjectId, subscriptionId);
String subscriptionId, String url, String serviceIdentityEmail) {
ProjectTopicName topicName = ProjectTopicName.of(tenantProjectId, topicId);
ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of(servicesProjectId, subscriptionId);
PushConfig config = PushConfig.newBuilder()
.setPushEndpoint(url)
.setOidcToken(PushConfig.OidcToken.newBuilder().setServiceAccountEmail(serviceIdentityEmail).build())
.build();
return this.subscriptionAdminClient.createSubscription(com.google.pubsub.v1.Subscription.newBuilder()
.setName(subscriptionName.toString())
.setTopic(topicName.toString())
.setPushConfig(config)
.setExpirationPolicy(ExpirationPolicy.newBuilder().setTtl(
Duration.newBuilder().setSeconds(SUBSCRIPTION_EXPIRATION_SECONDS).build()).build())
.setMessageRetentionDuration(Duration.newBuilder().setSeconds(
MESSAGE_RETENTION_SECONDS).build())
.setAckDeadlineSeconds(ACK_DEADLINE_SECONDS)
.build());
}
PushConfig config = PushConfig.newBuilder()
.setPushEndpoint(url)
.setOidcToken(PushConfig.OidcToken.newBuilder().setServiceAccountEmail(serviceIdentityEmail).build())
.build();
return subscriptionAdminClient.createSubscription(com.google.pubsub.v1.Subscription.newBuilder()
.setName(subscriptionName.toString())
.setTopic(topicName.toString())
.setPushConfig(config)
.setExpirationPolicy(ExpirationPolicy.newBuilder().setTtl(
Duration.newBuilder().setSeconds(SUBSCRIPTION_EXPIRATION_SECONDS).build()).build())
.setMessageRetentionDuration(Duration.newBuilder().setSeconds(
MESSAGE_RETENTION_SECONDS).build())
.setAckDeadlineSeconds(ACK_DEADLINE_SECONDS)
.build());
@PreDestroy
public void shutDownSubscriptionAdminClient() throws InterruptedException {
if (Objects.nonNull(this.subscriptionAdminClient)) {
log.debug("Shutting down publisher on exit, await termination duration: {} seconds", AWAIT_TERMINATION_DURATION);
subscriptionAdminClient.shutdown();
subscriptionAdminClient.awaitTermination(AWAIT_TERMINATION_DURATION, TimeUnit.SECONDS);
}
}
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment