test(firestore): fix flaky BulkWriter thread tests

Assert on individual thread pool executor configuration parameters
(max_length) instead of counting the global VM thread list size.

This fixes flakiness caused by:
- Global active VM thread count instability.
- Timing race conditions on thread execution lifetimes.

Please note that the original test was checking the sum of both thread pools, and now we are checking each pool individually and directly

Author: yoshi-automation

google-cloud-firestore/test/google/cloud/firestore/bulkwriter_test.rb

@@ -209,17 +209,20 @@
       stub = BatchWriteStub.new responses, requests, 1
       firestore.service.instance_variable_set :@firestore, stub
 
-      thread_count = Thread.list.count
       bw = firestore.bulk_writer request_threads: 1, batch_threads: 2
 
       result_1 = bw.create "cities/NYC", { foo: "bar"}
       result_2 = bw.create "cities/MTV", { foo: "bar"}
       result_3 = bw.create "cities/KIR", { foo: "bar"}
       bw.flush
-      thread_count_2 = Thread.list.count
+
+      write_tp = bw.instance_variable_get(:@write_thread_pool)
+      batch_tp = bw.instance_variable_get(:@scheduler).instance_variable_get(:@batch_thread_pool)
+      _(write_tp.max_length).must_equal 1
+      _(batch_tp.max_length).must_equal 2
+
       bw.close
 
-      _(thread_count_2 - thread_count).must_be :==, 3
       _(result_1.value).must_be_kind_of Google::Cloud::Firestore::BulkWriterOperation::WriteResult
       _(result_2.value).must_be_kind_of Google::Cloud::Firestore::BulkWriterOperation::WriteResult
       _(result_3.value).must_be_kind_of Google::Cloud::Firestore::BulkWriterOperation::WriteResult
@@ -231,18 +234,19 @@
       stub = BatchWriteStub.new [], [], 1
       firestore.service.instance_variable_set :@firestore, stub
 
-      thread_count = Thread.list.count
       bw = firestore.bulk_writer batch_threads: 3, request_threads: 1, retries: 1
 
       (1..100).each do
         bw.create "cities/#{SecureRandom.hex(4)}", { foo: "bar"}
       end
       bw.flush
-      thread_count_2 = Thread.list.count
 
-      bw.close
+      write_tp = bw.instance_variable_get(:@write_thread_pool)
+      batch_tp = bw.instance_variable_get(:@scheduler).instance_variable_get(:@batch_thread_pool)
+      _(write_tp.max_length).must_equal 1
+      _(batch_tp.max_length).must_equal 3
 
-      _(thread_count_2 - thread_count).must_be :==, 4
+      bw.close
     end
   end