[python-package][dask] handle failures parsing worker host names (#4852)

* [python-package][dask] handle failures parsing work host names

* add tests

* revert local testing changes

python-package/lightgbm/dask.py

@@ -95,6 +95,8 @@ def _group_workers_by_host(worker_addresses: Iterable[str]) -> Dict[str, _HostWo
     host_to_workers: Dict[str, _HostWorkers] = {}
     for address in worker_addresses:
         hostname = urlparse(address).hostname
+        if not hostname:
+            raise ValueError(f"Could not parse host name from worker address '{address}'")
         if hostname not in host_to_workers:
             host_to_workers[hostname] = _HostWorkers(default=address, all=[address])
         else:
@@ -380,6 +382,8 @@ def _machines_to_worker_map(machines: str, worker_addresses: Iterable[str]) -> D
     out = {}
     for address in worker_addresses:
         worker_host = urlparse(address).hostname
+        if not worker_host:
+            raise ValueError(f"Could not parse host name from worker address '{address}'")
         out[address] = machine_to_port[worker_host].pop()
 
     return out

tests/python_package_test/test_dask.py

@@ -469,6 +469,19 @@ def test_group_workers_by_host():
     assert host_to_workers == expected
 
 
+def test_group_workers_by_host_unparseable_host_names():
+    workers_without_protocol = ['0.0.0.1:80', '0.0.0.2:80']
+    with pytest.raises(ValueError, match="Could not parse host name from worker address '0.0.0.1:80'"):
+        lgb.dask._group_workers_by_host(workers_without_protocol)
+
+
+def test_machines_to_worker_map_unparseable_host_names():
+    workers = {'0.0.0.1:80': {}, '0.0.0.2:80': {}}
+    machines = "0.0.0.1:80,0.0.0.2:80"
+    with pytest.raises(ValueError, match="Could not parse host name from worker address '0.0.0.1:80'"):
+        lgb.dask._machines_to_worker_map(machines=machines, worker_addresses=workers.keys())
+
+
 def test_assign_open_ports_to_workers(cluster):
     with Client(cluster) as client:
         workers = client.scheduler_info()['workers'].keys()