Let me open a thread, its a valid suggestion, will attach the discussions Thanks, Susmit
On Wed, May 7, 2025 at 8:34 PM Julian Hyde <jhyde.apa...@gmail.com> wrote: > This thread is getting rather long. I think you should open a case. That’s > a permanent place that you can attach screenshots, stack traces. And it > will also be found by anyone who has the same problem in future. > > Julian > > > On May 7, 2025, at 2:51 AM, David Li <lidav...@apache.org> wrote: > > > > Thanks for reporting back. (Note, the attachment didn't make it.) Hmm, > I'd probably need to poke at it in a debugger. But it sounds like the > callbacks are getting lost somehow. (I'm not a fan of how the Java API > tries to adapt an async API into a synchronous one, not least because it > leads to weird issues like this.) > > > >> On Wed, May 7, 2025, at 14:42, Susmit Sarkar wrote: > >> Hi David > >> > >> 1. the python server thinks the RPC is complete: Yes > >> 2. the Java client got any (or all) of the data before getting stuck? > Depends, if we are reading in batches of 1000 which is pretty low number we > see data being streamed till 15000-18000 rows ultimately stuck, if we > select a batch size of 100K client is stuck indefinitely > >> 3. Sharing the debug screenshot , the method was invoked as expected > >> > >> image.png > >> > >> The issue is only seen when the client runs on a local machine > irrespective of the OS (both windows and Mac). We don't see the issue if > the same client code runs in a VM server pointing to the Arrow flight > server in another VM > >> Thanks, > >> Susmit > >> > >>> On Sat, May 3, 2025 at 12:17 PM David Li <lidav...@apache.org> wrote: > >>> Thanks Susmit. I can't quite see a bug but I'm willing to believe that > Flight is buggy in Java (I have never understood why the code tries to > manually handle flow control and without my memories of 6 years ago the > code seems suspect to me now) > >>> > >>> Do you know if (1) the python server thinks the RPC is complete, (2) > the Java client got any (or all) of the data before getting stuck? It may > also be interesting to step though the Java code with a debugger attached, > and see what the values of `pending` and `completed` are in the > FlightStream instance, and if the methods here[1] are all being hit as > expected. > >>> > >>> [1] > https://github.com/apache/arrow-java/blob/b9e37f0ccecc2651fec3487472c203bd223290e8/flight/flight-core/src/main/java/org/apache/arrow/flight/FlightClient.java#L356 > >>> > >>> On Fri, May 2, 2025, at 14:56, Susmit Sarkar wrote: > >>>> Hi James, > >>>> > >>>> The flight server is written in python. Here is the below code > snippet. > >>>> > >>>> We have overridden the *do_get* method only > >>>> > >>>> getStream() and getFlightInfo() is left untouched / default > implementation. > >>>> > >>>> def do_get(self, context, ticket): > >>>> """ > >>>> Handles client requests for data. The ticket will contain: > >>>> - access_key: S3 access key > >>>> - secret_key: S3 secret key > >>>> - s3_path: Full S3 path (e.g., bucket_name/object_key) > >>>> - mode: 'batch' (for batch streaming) or 'full' (for loading > >>>> the entire dataset) > >>>> """ > >>>> try: > >>>> # Parse the ticket to extract credentials, S3 path, and mode > >>>> access_key, secret_key, s3_path, mode, batch_size = > parse_ticket(ticket) > >>>> except InvalidTicketFormatError as e: > >>>> logging.error(str(e)) > >>>> raise > >>>> except InvalidModeError as e: > >>>> logging.error(str(e)) > >>>> raise > >>>> > >>>> # s3fs dont need s3a protocol. > >>>> if s3_path.startswith("s3://"): > >>>> s3_path = s3_path.replace("s3://", "", 1) > >>>> > >>>> logging.info(f"Cloudian S3 Override endpoint: > >>>> {Config.S3_ENDPOINT_OVERRIDE}") > >>>> logging.info(f"Cloudian S3 Region: {Config.S3_REGION}") > >>>> logging.info(f"Fetching Parquet data from S3: {s3_path} in mode: > {mode}") > >>>> > >>>> # Initialize the S3 handler with credentials > >>>> try: > >>>> s3_handler = S3Handler( > >>>> endpoint=Config.S3_ENDPOINT_OVERRIDE, > >>>> region=Config.S3_REGION, > >>>> access_key=access_key, > >>>> secret_key=secret_key > >>>> ) > >>>> except Exception as e: > >>>> logging.error(f"Error initializing S3 handler: {str(e)}") > >>>> raise S3AccessError(f"Error initializing S3 handler: > {str(e)}") from e > >>>> > >>>> if mode == DataRetrievalMode.BATCH: > >>>> try: > >>>> # Use the get_parquet_data method for both wildcard and > >>>> non-wildcard cases > >>>> parquet_data = s3_handler.get_parquet_data(s3_path) > >>>> # parquet_data.schema: This is used when parquet_data is > >>>> an instance of ds.Dataset > >>>> # (i.e., when multiple Parquet files are being processed > >>>> as a dataset). > >>>> # > >>>> # parquet_data.schema_arrow: This is used when > >>>> parquet_data is an instance of pq (pyarrow.parquet) module. > >>>> # A single Parquet file has its own schema, accessible > >>>> via schema_arrow in PyArrow > >>>> schema = parquet_data.schema if isinstance(parquet_data, > >>>> ds.Dataset) else parquet_data.schema_arrow > >>>> return flight.GeneratorStream(schema, > >>>> s3_handler.stream_parquet_batches(parquet_data, batch_size)) > >>>> except OSError as e: > >>>> logging.error(f"AWS S3 access error: {str(e)}") > >>>> raise S3AccessError(f"Failed to access S3: {str(e)}") from > e > >>>> except Exception as e: > >>>> logging.error(f"Error streaming Parquet data: {str(e)}") > >>>> raise DataProcessingError(f"Error streaming Parquet data: > >>>> {str(e)}") from e > >>>> > >>>> # Handle 'full' mode to load the entire dataset > >>>> elif mode == DataRetrievalMode.FULL: > >>>> try: > >>>> # Check if the S3 path contains a wildcard and the mode is > FULL > >>>> if "*" in s3_path: > >>>> logging.warning( > >>>> f"Wildcard pattern detected in S3 path '{s3_path}' > >>>> with FULL data retrieval mode. " > >>>> f"This may put pressure on memory as all files > >>>> will be loaded into memory at once." > >>>> ) > >>>> # Use the same get_parquet_data method for both wildcard > >>>> and non-wildcard cases > >>>> parquet_data = s3_handler.get_parquet_data(s3_path) > >>>> # Load the entire dataset into memory / Chance of OOM. > >>>> # table = parquet_data.to_table() if > >>>> isinstance(parquet_data, ds.Dataset) else parquet_data.read_table() > >>>> # Load the entire dataset into memory, with consideration > >>>> for Dataset vs. ParquetFile > >>>> if isinstance(parquet_data, ds.Dataset): > >>>> table = parquet_data.to_table() > >>>> else: > >>>> table = parquet_data.read() > >>>> return flight.RecordBatchStream(table) > >>>> except OSError as e: > >>>> logging.error(f"AWS S3 access error: {str(e)}") > >>>> raise S3AccessError(f"Failed to access S3: {str(e)}") from > e > >>>> except Exception as e: > >>>> logging.error(f"Error loading full Parquet dataset: > {str(e)}") > >>>> raise DataProcessingError(f"Error loading full Parquet > >>>> dataset: {str(e)}") from e > >>>> > >>>> else: > >>>> logging.error(f"Invalid mode: > >>>> {DataRetrievalMode.from_string(mode)}. Expected 'batch' or 'full'.") > >>>> raise InvalidModeError() > >>>> > >>>> > >>>> // Helper functions. > >>>> > >>>> def get_parquet_data(self, s3_path): > >>>> """ > >>>> Retrieves Parquet data from S3. If the path contains a > >>>> wildcard pattern, it lists all matching files manually. > >>>> If it's a single file, it reads the file directly. > >>>> > >>>> :param s3_path: The S3 path, which could be a wildcard pattern > >>>> or a direct file path. > >>>> :return: PyArrow Dataset object if it's a wildcard, or a > >>>> ParquetFile object for a single file. > >>>> """ > >>>> try: > >>>> # Check if the path contains a wildcard > >>>> if "*" in s3_path: > >>>> # Split the directory and pattern (e.g., `*.parquet`) > >>>> directory, pattern = s3_path.rsplit("/", 1) > >>>> > >>>> # List all files in the directory and filter using the > pattern > >>>> logging.info(f"Fetching Parquet files matching wildcard: > {s3_path}") > >>>> files = > self.s3_fs.get_file_info(fs.FileSelector(directory)) > >>>> > >>>> # Filter files matching the pattern (e.g., *.parquet) and > >>>> sort them by modification time (mtime_ns) > >>>> sorted_file_paths = [file.path for file in sorted(files, > >>>> key=lambda file: file.mtime_ns) if fnmatch.fnmatch(file.path, > >>>> f"{directory}/{pattern}")] > >>>> > >>>> if not sorted_file_paths: > >>>> raise FileNotFoundError(f"No files matching pattern > >>>> {pattern} found in {directory}") > >>>> > >>>> logging.info(f"Sorted files: {sorted_file_paths}") > >>>> > >>>> # Validate schemas across all files > >>>> if not validate_schemas(sorted_file_paths, self.s3_fs): > >>>> raise ValueError("Schema mismatch detected across > files.") > >>>> > >>>> # Create a dataset from the matching files > >>>> dataset = ds.dataset(sorted_file_paths, format="parquet", > >>>> filesystem=self.s3_fs) > >>>> return dataset > >>>> else: > >>>> # Handle single file case: read the specific Parquet file > >>>> logging.info(f"Fetching single Parquet file: {s3_path}") > >>>> parquet_file = > pq.ParquetFile(self.s3_fs.open_input_file(s3_path)) > >>>> return parquet_file > >>>> except Exception as e: > >>>> logging.error(f"Error fetching Parquet data from S3: {e}") > >>>> raise e > >>>> > >>>> @staticmethod > >>>> def stream_parquet_batches(parquet_data, batch_size=None): > >>>> """ > >>>> Stream the Parquet data in batches. Supports both datasets > >>>> (multiple files) and single Parquet files. > >>>> > >>>> :param parquet_data: The Dataset or ParquetFile object to > >>>> stream data from. > >>>> :param batch_size: The size of the batches to stream. Default > >>>> is 100,000 if not provided. > >>>> :return: Generator for streaming Parquet batches. > >>>> """ > >>>> try: > >>>> # Ensure batch_size is an integer, set default if None > >>>> if batch_size is None or not isinstance(batch_size, int): > >>>> batch_size = 100000 > >>>> > >>>> if isinstance(parquet_data, ds.Dataset): > >>>> # If it's a dataset (multiple files), stream dataset > >>>> batches using `int_batch_size` > >>>> logging.info(f"Streaming Parquet data in batches from a > dataset") > >>>> for batch in > parquet_data.to_batches(batch_size=batch_size): > >>>> yield batch > >>>> else: > >>>> # If it's a single file (ParquetFile), stream file batches > >>>> (iter_batches) > >>>> logging.info(f"Streaming Parquet data in batches from a > >>>> single file") > >>>> for batch in > parquet_data.iter_batches(batch_size=batch_size): > >>>> yield batch > >>>> except Exception as e: > >>>> logging.error(f"Error streaming Parquet batches: {e}") > >>>> raise e > >>>> > >>>> > >>>> On Wed, Apr 30, 2025 at 11:14 PM James Duong > >>>> <james.du...@improving.com.invalid> wrote: > >>>> > >>>>> Would you be able to share the server’s getStream() and > getFlightInfo() > >>>>> implementations? > >>>>> > >>>>> Note that getStream() needs should be written such that it doesn’t > block > >>>>> the grpc thread. > >>>>> > >>>>> > >>>>> Get Outlook for Mac <https://aka.ms/GetOutlookForMac> > >>>>> > >>>>> From: Susmit Sarkar <susmitsir...@gmail.com> > >>>>> Date: Wednesday, April 30, 2025 at 2:59 AM > >>>>> To: David Li <lidav...@apache.org> > >>>>> Cc: nik.9...@gmail.com <nik.9...@gmail.com>, dev@arrow.apache.org < > >>>>> dev@arrow.apache.org> > >>>>> Subject: Re: Query on stuck Arrow Flight Client while interacting > from > >>>>> local workstation (mac) > >>>>> > >>>>> Hi David > >>>>> > >>>>> Sharing the arrow client thread dump for reference. Strangely if we > pass a > >>>>> dummy non existent s3 path we are getting proper error from server > >>>>> > >>>>> cef_flight_server.exceptions.S3AccessError: Failed to access S3: > [Errno 2] > >>>>> Path does not exist > >>>>> > 'bg0975-cef-ccmedev-data/pp/load_date=2024-11-21/part-00007.c008.snappy.parquet'. > >>>>> Detail: [errno 2] No such file or directory > >>>>> > >>>>> Which translates the server is reachable and we do see the logs in > server > >>>>> as well > >>>>> > >>>>> It works fine if we call the client within a VM issue arises in local > >>>>> workstation, where its stuck indefinitely. > >>>>> > >>>>> Thanks, > >>>>> Susmit > >>>>> > >>>>> On Wed, Apr 30, 2025 at 12:54 PM David Li <lidav...@apache.org > <mailto: > >>>>> lidav...@apache.org>> wrote: > >>>>> This is not specific to Flight; use jstack or your favorite > >>>>> instrumentation tool (VisualVM etc.) > >>>>> > >>>>> On Wed, Apr 30, 2025, at 15:41, NIKHIL RANJAN wrote: > >>>>>> Hi David, > >>>>>> > >>>>>> How to enable thread dump logs for both client and server code. > >>>>>> > >>>>>> As of now, I don't see any error on either client side or server > side. It > >>>>>> just hangs/gets stuck. > >>>>>> > >>>>>> Thanks, > >>>>>> Nikhil > >>>>>> > >>>>>> On Thu, 24 Apr, 2025, 14:39 Susmit Sarkar, <susmitsir...@gmail.com > >>>>> <mailto:susmitsir...@gmail.com>> wrote: > >>>>>> > >>>>>>> Hi Team, > >>>>>>> > >>>>>>> We are using this below code snippet in Scala to query the flight > >>>>> server, > >>>>>>> but seems to be stuck indefinitely, this issue is seen when we are > >>>>> testing > >>>>>>> from our local workstation (Mac to be precise) > >>>>>>> > >>>>>>> Another interesting thing, it's able to propagate the error message > >>>>>>> correctly but not the FlightStream data, the same code works fine > when > >>>>> we > >>>>>>> run inside a linux VM. > >>>>>>> > >>>>>>> Do you folks see any issue in the code? > >>>>>>> > >>>>>>> def fetchDataStreamIterator(details: BaseDataAccessDetails): > >>>>> Iterator[FlightStream] = { > >>>>>>> logger.info<http://logger.info>(s"Fetching data for details: > >>>>> ${details.toString}") > >>>>>>> val ticketStr = buildTicketStr(details) > >>>>>>> logger.info<http://logger.info>(s"Generated ticket string: > >>>>> $ticketStr") > >>>>>>> > >>>>>>> val allocator = new RootAllocator(Long.MaxValue) > >>>>>>> val client = FlightClient.builder(allocator, > >>>>> Location.forGrpcInsecure(serverHost, serverPort)).build() > >>>>>>> > >>>>>>> try { > >>>>>>> val ticket = new > Ticket(ticketStr.getBytes(StandardCharsets.UTF_8)) > >>>>>>> val stream = client.getStream(ticket) > >>>>>>> > >>>>>>> Iterator.continually { > >>>>>>> if (stream.next()) Some(stream) else { > >>>>>>> // Cleanup when no more batches > >>>>>>> close(stream, client) > >>>>>>> None > >>>>>>> } > >>>>>>> }.takeWhile(_.isDefined).flatten > >>>>>>> } catch { > >>>>>>> case e: FlightRuntimeException => > >>>>>>> logger.error(s"Error communicating with Flight server: > >>>>> ${e.getMessage}") > >>>>>>> throw new CefFlightServerException(s"Error communicating with > >>>>> Flight server: ${e.getMessage}", e) > >>>>>>> case e: Exception => > >>>>>>> logger.error(s"Failed to fetch data: ${e.getMessage}") > >>>>>>> throw new CefFlightServerException("Failed to fetch data from > >>>>> Flight Server", e) > >>>>>>> } > >>>>>>> } > >>>>>>> > >>>>>>> > >>>>>>> Thanks, > >>>>>>> > >>>>>>> Susmit > >>>>>>> > >>>>>>> > >>>>> > >>>>> Warning: The sender of this message could not be validated and may > not be > >>>>> the actual sender. > >>>>> >
