finally:
    query.stop()
    print(f"query_active_after_stop={query.isActive}")
    print(f"active_streams_after_stop={len(spark.streams.active)}")
    spark.stop()
    shutil.rmtree(BASE_DIR, ignore_errors=True)
    print(f"cleanup_exists_after_stop={BASE_DIR.exists()}")