(self)
| 665 | return self._config.session.execution_mode == "eager" |
| 666 | |
| 667 | def _send_heartbeat(self): |
| 668 | # >1: failure, 0: reset when success |
| 669 | heartbeat_failure_count = 0 |
| 670 | while not self._closed: |
| 671 | if self._grpc_client: |
| 672 | try: |
| 673 | self._grpc_client.send_heartbeat() |
| 674 | except Exception as exc: |
| 675 | if heartbeat_failure_count == 0: |
| 676 | logger.warning("Failed to send heartbeat message", exc_info=exc) |
| 677 | heartbeat_failure_count = heartbeat_failure_count + 1 |
| 678 | if heartbeat_failure_count > self._heartbeat_maximum_failures: |
| 679 | logger.error( |
| 680 | "The connection between coordinator has lost after %d times " |
| 681 | "of heartbeat failure, closing the session ...", |
| 682 | heartbeat_failure_count, |
| 683 | ) |
| 684 | self.close() |
| 685 | self._disconnected = True |
| 686 | else: |
| 687 | heartbeat_failure_count = 0 |
| 688 | self._disconnected = False |
| 689 | time.sleep(self._heartbeat_interval_seconds) |
| 690 | |
| 691 | def connected(self) -> bool: |
| 692 | """Check if the session is still connected and available. |
nothing calls this directly
no test coverage detected