WIP: DataStreams processor and Kafka consumer updates

Author: ericfirth

lib/datadog/tracing/contrib/kafka/instrumentation/consumer.rb

@@ -20,13 +20,13 @@ def each_message(**kwargs, &block)
                   proc do |message|
                     # DSM: Create checkpoint for consumed message
                     Datadog.logger.debug { "Kafka each_message: DSM enabled for topic #{message.topic}" }
-                    
+
                     processor = Datadog.configuration.tracing.data_streams.processor
-                    
+
                     # Extract pathway context from message headers if available
                     headers = message.headers || {}
                     processor.set_consume_checkpoint('kafka', message.topic) { |key| headers[key] }
-                    
+
                     # Call the original block if provided
                     block.call(message) if block_given?
                   end
@@ -45,13 +45,13 @@ def each_batch(**kwargs, &block)
                   proc do |batch|
                     # DSM: Create checkpoint for consumed batch
                     Datadog.logger.debug { "Kafka each_batch: DSM enabled for topic #{batch.topic}" }
-                    
+
                     processor = Datadog.configuration.tracing.data_streams.processor
-                    
+
                     # For batch processing, we don't have individual message headers
                     # so we create a consume checkpoint without pathway context
                     processor.set_consume_checkpoint('kafka', batch.topic)
-                    
+
                     # Call the original block if provided
                     block.call(batch) if block_given?
                   end

spec/datadog/tracing/data_streams/processor_spec.rb

@@ -187,25 +187,25 @@
       it 'creates produce checkpoint with correct tags' do
         carrier_set = double('carrier_set')
         expect(carrier_set).to receive(:call).with('dd-pathway-ctx-base64', anything)
-        
+
         result = processor.set_produce_checkpoint('kafka', 'orders-topic') { |key, value| carrier_set.call(key, value) }
-        
+
         expect(result).to be_a(String)
         expect(result).not_to be_empty
       end
 
       it 'works without block' do
         result = processor.set_produce_checkpoint('kafka', 'orders-topic')
-        
+
         expect(result).to be_a(String)
         expect(result).not_to be_empty
       end
 
       it 'returns nil when processor disabled' do
         processor.enabled = false
-        
+
         result = processor.set_produce_checkpoint('kafka', 'orders-topic')
-        
+
         expect(result).to be_nil
       end
     end
@@ -214,9 +214,9 @@
       it 'creates consume checkpoint with correct tags' do
         carrier_get = double('carrier_get')
         expect(carrier_get).to receive(:call).with('dd-pathway-ctx-base64').and_return(nil)
-        
+
         result = processor.set_consume_checkpoint('kafka', 'orders-topic') { |key| carrier_get.call(key) }
-        
+
         expect(result).to be_a(String)
         expect(result).not_to be_empty
       end
@@ -226,29 +226,29 @@
         carrier_get = double('carrier_get')
         expect(carrier_get).to receive(:call).with('dd-pathway-ctx-base64').and_return(encoded_context)
         expect(processor).to receive(:decode_pathway_b64).with(encoded_context)
-        
+
         processor.set_consume_checkpoint('kafka', 'orders-topic') { |key| carrier_get.call(key) }
       end
 
       it 'works without block' do
         result = processor.set_consume_checkpoint('kafka', 'orders-topic')
-        
+
         expect(result).to be_a(String)
         expect(result).not_to be_empty
       end
 
       it 'respects manual_checkpoint parameter' do
         result = processor.set_consume_checkpoint('kafka', 'orders-topic', manual_checkpoint: false)
-        
+
         expect(result).to be_a(String)
         expect(result).not_to be_empty
       end
 
       it 'returns nil when processor disabled' do
         processor.enabled = false
-        
+
         result = processor.set_consume_checkpoint('kafka', 'orders-topic')
-        
+
         expect(result).to be_nil
       end
     end