o-nikolas commented on code in PR #23490:
URL: https://github.com/apache/airflow/pull/23490#discussion_r866122974
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -368,20 +368,25 @@ def extract_xcom(self, pod: V1Pod) -> str:
_preload_content=False,
)
) as resp:
- result = self._exec_pod_command(resp, f'cat
{PodDefaults.XCOM_MOUNT_PATH}/return.json')
+ result = self._exec_pod_command(resp, f'cat
{PodDefaults.XCOM_MOUNT_PATH}/return.json', True)
self._exec_pod_command(resp, 'kill -s SIGINT 1')
if result is None:
raise AirflowException(f'Failed to extract xcom from pod:
{pod.metadata.name}')
return result
- def _exec_pod_command(self, resp, command: str) -> Optional[str]:
+ def _exec_pod_command(self, resp, command: str,
extract_full_content=False) -> Optional[str]:
if resp.is_open():
self.log.info('Running command... %s\n', command)
resp.write_stdin(command + '\n')
while resp.is_open():
resp.update(timeout=1)
+ res = None
if resp.peek_stdout():
- return resp.read_stdout()
+ res = resp.read_stdout()
+ if extract_full_content:
Review Comment:
What is the value in keeping the previous broken behaviour? I'd say we just
keep the while loop for all cases, I don't see why you'd want partial
extraction especially if there is a risk of it throwing an exception when it is
json loaded.
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -368,20 +368,25 @@ def extract_xcom(self, pod: V1Pod) -> str:
_preload_content=False,
)
) as resp:
- result = self._exec_pod_command(resp, f'cat
{PodDefaults.XCOM_MOUNT_PATH}/return.json')
+ result = self._exec_pod_command(resp, f'cat
{PodDefaults.XCOM_MOUNT_PATH}/return.json', True)
self._exec_pod_command(resp, 'kill -s SIGINT 1')
if result is None:
raise AirflowException(f'Failed to extract xcom from pod:
{pod.metadata.name}')
return result
- def _exec_pod_command(self, resp, command: str) -> Optional[str]:
+ def _exec_pod_command(self, resp, command: str,
extract_full_content=False) -> Optional[str]:
if resp.is_open():
self.log.info('Running command... %s\n', command)
resp.write_stdin(command + '\n')
while resp.is_open():
resp.update(timeout=1)
+ res = None
if resp.peek_stdout():
- return resp.read_stdout()
+ res = resp.read_stdout()
+ if extract_full_content:
+ while resp.peek_stdout():
+ res = res + resp.read_stdout()
+ return res
if resp.peek_stderr():
self.log.info("stderr from command: %s",
resp.read_stderr())
Review Comment:
The same fix should be applied to reading from std_err, no?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
