Rescue ThreadError when creating Sender Thread

lib/datadog/statsd/sender.rb

@@ -20,6 +20,7 @@ def initialize(message_buffer, telemetry: nil, queue_size: UDP_DEFAULT_BUFFER_SI
         @mx = Mutex.new
         @queue_class = queue_class
         @thread_class = thread_class
+        @done = false
         @flush_timer = if flush_interval
           Datadog::Statsd::Timer.new(flush_interval) { flush(sync: true) }
         else
@@ -66,8 +67,11 @@ def add(message)
         # if the thread does not exist, we assume we are running in a forked process,
         # empty the message queue and message buffers (these messages belong to
         # the parent process) and spawn a new companion thread.
-        if !sender_thread.alive?
+        if sender_thread.nil? || !sender_thread.alive?
           @mx.synchronize {
+            # an attempt was previously made to start the sender thread but failed.
+            # skipping re-start
+            return if @done
             # a call from another thread has already re-created
             # the companion thread before this one acquired the lock
             break if sender_thread.alive?
@@ -96,9 +100,15 @@ def start
 
         # initialize a new message queue for the background thread
         @message_queue = @queue_class.new
-        # start background thread
-        @sender_thread = @thread_class.new(&method(:send_loop))
-        @sender_thread.name = "Statsd Sender" unless Gem::Version.new(RUBY_VERSION) < Gem::Version.new('2.3')
+        begin
+          # start background thread
+          @sender_thread = @thread_class.new(&method(:send_loop))
+          @sender_thread.name = "Statsd Sender" unless Gem::Version.new(RUBY_VERSION) < Gem::Version.new('2.3')
+        rescue ThreadError => e
+          @logger.debug { "Statsd: Failed to start sender thread: #{e.message}" } if @logger
+          @mx.synchronize { @done = true }
+        end
+
         @flush_timer.start if @flush_timer
       end
 

spec/statsd/sender_spec.rb

@@ -62,6 +62,45 @@
       end
     end
 
+    context 'when a ThreadError is raised starting the sender thread' do
+      let(:thread_class) do
+        class_double(Thread)
+      end
+
+      before do
+        expect(thread_class).to receive(:new).and_raise(ThreadError, "ThreadError")
+      end
+
+      it 'ignores the thread error' do
+        expect do
+          subject.start
+        end.not_to raise_error
+      end
+
+      context 'when add' do
+        let(:fake_queue_length) { queue_size }
+        let(:fake_queue) do
+          if Queue.instance_methods.include?(:close)
+            instance_double(Queue, { "length" => fake_queue_length, "<<" => true, "close" => true })
+          else
+            instance_double(Queue, { "length" => fake_queue_length, "<<" => true })
+          end
+        end
+        let(:queue_class) do
+          class_double(Queue, new: fake_queue)
+        end
+
+        it 'does not store the message' do
+          subject.start
+          expect(fake_queue).not_to receive(:<<).with('message')
+          if not Queue.instance_methods.include?(:close)
+            expect(fake_queue).to receive(:<<).with(:close)
+          end
+          subject.add('message')
+        end
+      end
+    end
+
     context 'when flush_interval is set' do
       let(:flush_interval) { 0.001 }