feat(fcm): Migrate topic management to FCM v1 API - #1240
lahirumaramba wants to merge 3 commits into
Conversation
fdfe8ea to
51005bd
Compare
There was a problem hiding this comment.
Code Review
This pull request migrates topic subscription and unsubscription operations in FirebaseMessaging from the legacy Instance ID API to the FCM v1 API, while retaining the legacy methods as deprecated. The new implementation in FirebaseMessagingClientImpl handles these operations concurrently using CompletableFuture. However, the current design creates and shuts down a new thread pool for every topic management request, which is highly inefficient and can cause thread exhaustion under high load. It is recommended to initialize a single shared ExecutorService in the constructor with core thread timeout enabled, and reuse it across requests to improve performance and simplify the request handling logic.
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request migrates topic subscription and unsubscription operations to the FCM v1 API, introducing new asynchronous methods in FirebaseMessagingClientImpl using CompletableFuture and deprecating the legacy Instance ID API methods. The feedback recommends using daemon threads in the default executor to prevent JVM shutdown delays in short-lived applications, and wrapping task submissions in a try-catch block to gracefully handle RejectedExecutionException when custom bounded executors are used.
d9277eb to
5511d97
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request migrates topic management operations (subscribe/unsubscribe) from the legacy Instance ID API to the FCM v1 API. It introduces a configurable FCM host, adds a dedicated executor service for concurrent topic management requests, and updates the internal client implementation and corresponding tests to support these changes. The review identified a potential issue where a trailing slash in a custom FCM host configuration could lead to malformed URLs, and a code suggestion was provided to normalize the host string.
| this.projectId = builder.projectId; | ||
| this.fcmHost = Strings.isNullOrEmpty(builder.fcmHost) ? DEFAULT_FCM_HOST : builder.fcmHost; | ||
| this.fcmSendUrl = String.format(FCM_URL, this.fcmHost, builder.projectId); |
There was a problem hiding this comment.
If builder.fcmHost is configured with a trailing slash (e.g., https://custom.host.com/), it can result in URLs with double slashes (e.g., https://custom.host.com//v1/projects/...). This can cause unexpected 404 errors or authentication failures with some strict API gateways or proxies. Consider normalizing fcmHost by stripping any trailing slash.
| this.projectId = builder.projectId; | |
| this.fcmHost = Strings.isNullOrEmpty(builder.fcmHost) ? DEFAULT_FCM_HOST : builder.fcmHost; | |
| this.fcmSendUrl = String.format(FCM_URL, this.fcmHost, builder.projectId); | |
| this.projectId = builder.projectId; | |
| String host = Strings.isNullOrEmpty(builder.fcmHost) ? DEFAULT_FCM_HOST : builder.fcmHost; | |
| if (host.endsWith("/")) { | |
| host = host.substring(0, host.length() - 1); | |
| } | |
| this.fcmHost = host; | |
| this.fcmSendUrl = String.format(FCM_URL, this.fcmHost, builder.projectId); |
jonathanedey
left a comment
There was a problem hiding this comment.
LGTM, but with some push back on the new thread pool.
| this.executor = builder.executor != null | ||
| ? builder.executor | ||
| : createDefaultExecutor(builder.threadFactory); | ||
| this.threadFactory = builder.threadFactory; | ||
| } | ||
|
|
||
| private static ExecutorService createDefaultExecutor(ThreadFactory threadFactory) { | ||
| ThreadFactory baseFactory = | ||
| threadFactory != null ? threadFactory : Executors.defaultThreadFactory(); | ||
| ThreadFactory factory = new ThreadFactoryBuilder() | ||
| .setThreadFactory(baseFactory) | ||
| .setNameFormat("firebase-messaging-topics-%d") | ||
| .setDaemon(true) | ||
| .build(); | ||
| ThreadPoolExecutor pool = new ThreadPoolExecutor( | ||
| 100, 100, | ||
| 60L, TimeUnit.SECONDS, | ||
| new LinkedBlockingQueue<Runnable>(), | ||
| factory); | ||
| pool.allowCoreThreadTimeOut(true); | ||
| return pool; |
There was a problem hiding this comment.
I'm hesitant on adding a nested thread executor here. The existing implementation already uses an app level thread manager which developers can configure through options. Adding another layer of threads without developer control could have be a negative side effect of more thread overhead than the developer is expecting.
| List<CompletableFuture<TopicResult>> futures = new ArrayList<>(); | ||
| for (int i = 0; i < registrationTokens.size(); i++) { | ||
| final int index = i; | ||
| final String token = registrationTokens.get(i); | ||
| try { | ||
| futures.add(CompletableFuture.supplyAsync( | ||
| () -> sendSingleTopicRequest(token, topicName, isSubscribe, index), | ||
| this.executor)); | ||
| } catch (RejectedExecutionException e) { | ||
| futures.add(CompletableFuture.completedFuture( | ||
| TopicResult.error(index, "REJECTED_BY_EXECUTOR"))); | ||
| } | ||
| } | ||
|
|
||
| CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); | ||
|
|
||
| int successCount = 0; | ||
| List<TopicManagementResponse.Error> errors = new ArrayList<>(); | ||
| for (CompletableFuture<TopicResult> future : futures) { | ||
| TopicResult result = future.join(); | ||
| if (result.isSuccess()) { | ||
| successCount++; | ||
| } else { | ||
| errors.add(new TopicManagementResponse.Error(result.getIndex(), result.getReason())); | ||
| } | ||
| } |
There was a problem hiding this comment.
I think we solved a similar issue without a new thread pool with sendEachAsync. Are there advantages to this method over the sendEachAsync way?
| private String extractReason(FirebaseMessagingException e) { | ||
| if (e.getMessagingErrorCode() != null) { | ||
| return e.getMessagingErrorCode().name(); | ||
| } | ||
| if (e.getHttpResponse() != null && !Strings.isNullOrEmpty(e.getHttpResponse().getContent())) { | ||
| try { | ||
| MessagingServiceErrorResponse parsed = jsonFactory.createJsonParser( | ||
| e.getHttpResponse().getContent()) | ||
| .parseAndClose(MessagingServiceErrorResponse.class); | ||
| if (parsed.getMessagingErrorCode() != null) { | ||
| return parsed.getMessagingErrorCode().name(); | ||
| } | ||
| if (!Strings.isNullOrEmpty(parsed.getStatus())) { | ||
| return parsed.getStatus(); | ||
| } | ||
| if (!Strings.isNullOrEmpty(parsed.getErrorMessage())) { | ||
| return parsed.getErrorMessage(); | ||
| } | ||
| } catch (Exception ignore) { | ||
| // Ignore JSON parsing errors | ||
| } | ||
| } | ||
| if (e.getErrorCode() != null && e.getErrorCode() != ErrorCode.UNKNOWN) { | ||
| return e.getErrorCode().name(); | ||
| } | ||
| if (e.getHttpResponse() != null) { | ||
| switch (e.getHttpResponse().getStatusCode()) { | ||
| case 400: | ||
| return "INVALID_ARGUMENT"; | ||
| case 401: | ||
| case 403: | ||
| return "PERMISSION_DENIED"; | ||
| case 404: | ||
| return "NOT_FOUND"; | ||
| case 429: | ||
| return "RESOURCE_EXHAUSTED"; | ||
| case 500: | ||
| return "INTERNAL"; | ||
| case 503: | ||
| return "DEADLINE_EXCEEDED"; | ||
| default: | ||
| return "UNKNOWN_ERROR"; | ||
| } | ||
| } | ||
| return "UNKNOWN_ERROR"; | ||
| } |
There was a problem hiding this comment.
Same as others languages. Not blocking but I don't think we need to reparse the firebase error. We can keep the custom mapping if nessasary for matching the status codes error codes correctly.
| case 500: | ||
| return "INTERNAL"; | ||
| case 503: | ||
| return "DEADLINE_EXCEEDED"; |
There was a problem hiding this comment.
is 503 "Service Unavailable" and 504 "DEADLINE_EXCEEDED"?
Migrates topic subscription and management from the legacy Instance ID (IID) service (
iid.googleapis.com) to the FCM v1 REST API (fcm.googleapis.com/v1/projects/{projectId}/registrations/{token}/topicSubscriptions).