Add close() method to ReorgAwareStream and remove unnecessary hasattr checks

- Add explicit close() method to ReorgAwareStream for proper stream cleanup
  on reorg restart (fixes resource leak)
- Remove hasattr check in client.py since both StreamingResultIterator and
  ReorgAwareStream have close()

Author: incrypto32

src/amp/client.py

@@ -838,9 +838,7 @@ def query_and_load_streaming(
 
                     # Check if we need to restart due to reorg
                     if reorg_result:
-                        # Close the old stream before restarting
-                        if hasattr(stream_iterator, 'close'):
-                            stream_iterator.close()
+                        stream_iterator.close()
                         self.logger.info('Reorg detected, restarting stream with new resume position...')
                         resume_watermark = loader_instance.state_store.get_resume_position(connection_name, destination)
                         continue

src/amp/streaming/reorg.py

@@ -176,12 +176,15 @@ def _is_duplicate_batch(self, current_ranges: List[BlockRange]) -> bool:
         # All ranges are exact duplicates
         return True
 
+    def close(self) -> None:
+        """Close the underlying stream"""
+        self.stream_iterator.close()
+
     def __enter__(self) -> 'ReorgAwareStream':
         """Context manager entry"""
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb) -> None:
         """Context manager exit"""
-        # Delegate to underlying stream
         if hasattr(self.stream_iterator, 'close'):
             self.stream_iterator.close()