feat(fcm): Migrate topic management to FCM v1 API

[lahirumaramba]

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).

[gemini-code-assist]

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.

[lahirumaramba]

/gemini review

[gemini-code-assist]

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.

[gemini-code-assist]

The default thread factory creates non-daemon threads. Since the thread pool has a 60-second keep-alive time and allows core thread timeouts, these non-daemon threads can prevent the JVM from exiting for up to 60 seconds after the main thread finishes. It is recommended to use daemon threads for the default executor to avoid hanging short-lived applications or CLI tools.

Suggested change

	private static ExecutorService createDefaultExecutor(ThreadFactory threadFactory) {
	ThreadPoolExecutor pool = new ThreadPoolExecutor(
	100, 100,
	60L, TimeUnit.SECONDS,
	new LinkedBlockingQueue<Runnable>(),
	threadFactory != null ? threadFactory : Executors.defaultThreadFactory());
	pool.allowCoreThreadTimeOut(true);
	return pool;
	}
	private static ExecutorService createDefaultExecutor(ThreadFactory threadFactory) {
	ThreadFactory factory = threadFactory != null ? threadFactory : new ThreadFactory() {
	private final ThreadFactory defaultFactory = Executors.defaultThreadFactory();
	@Override
	public Thread newThread(Runnable r) {
	Thread t = defaultFactory.newThread(r);
	t.setDaemon(true);
	t.setName("firebase-messaging-topic-mgt-" + t.getId());
	return t;
	}
	};
	ThreadPoolExecutor pool = new ThreadPoolExecutor(
	100, 100,
	60L, TimeUnit.SECONDS,
	new LinkedBlockingQueue<Runnable>(),
	factory);
	pool.allowCoreThreadTimeOut(true);
	return pool;
	}

[gemini-code-assist]

If a user configures a custom executor with a bounded queue and a rejection policy, calling CompletableFuture.supplyAsync can throw a RejectedExecutionException. If this happens, the entire batch operation will fail abruptly. Wrapping the submission in a try-catch block and returning a failed TopicResult for rejected tasks ensures defensive programming and allows the rest of the batch to complete or fail gracefully.

Suggested change

	List<CompletableFuture<TopicResult>> futures = new ArrayList<>();
	for (int i = 0; i < registrationTokens.size(); i++) {
	final int index = i;
	final String token = registrationTokens.get(i);
	futures.add(CompletableFuture.supplyAsync(
	() -> sendSingleTopicRequest(token, topicName, isSubscribe, index),
	this.executor));
	}
	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 (java.util.concurrent.RejectedExecutionException e) {
	futures.add(CompletableFuture.completedFuture(
	TopicResult.error(index, "REJECTED_BY_EXECUTOR")));
	}
	}