public BigQueryReadClient getBigQueryReadClient() throws Exception
	{
		final BigQueryReadSettings bigQueryReadSettings = BigQueryReadSettings.newBuilder()
				.setCredentialsProvider(
						FixedCredentialsProvider.create(credentials)
				)
//				.setTransportChannelProvider(channelProvider)
//				.setExecutorProvider(executorProvider)
				.build();

		final BigQueryReadClient readClient = BigQueryReadClient.create(bigQueryReadSettings);

		return readClient;
	}

        @NotNull
	private Flux<T> getTableRecords(long sysSkip, TableId destinationTable, Long sysTakeLong) throws Exception
	{
		final BigQueryReadClient readClient = bigQueryConfig.getBigQueryReadClient();

		final String table =
				String.format(
						"projects/%s/datasets/%s/tables/%s",
						bigQueryConfig.getProjectId(), destinationTable.getDataset(), destinationTable.getTable());
		final String parent = String.format("projects/%s", bigQueryConfig.getProjectId());

		// Prepare the read options
		final ReadSession.TableReadOptions tableReadOptions = api.getFields().keySet().stream()
				.reduce(
						ReadSession.TableReadOptions.newBuilder(),
						ReadSession.TableReadOptions.Builder::addSelectedFields,
						(b1, b2) -> b1
				)
				.setRowRestriction(sysTakeLong == null ?
						"row_number > " + sysSkip :
						"row_number BETWEEN " + sysSkip + " AND " + (sysSkip + sysTakeLong))
				.build();
		final ReadSession.Builder readSessionBuilder =
				ReadSession.newBuilder()
						.setTable(table)
						.setDataFormat(DataFormat.ARROW)
						.setReadOptions(tableReadOptions);
		final CreateReadSessionRequest createReadSessionRequest = CreateReadSessionRequest.newBuilder()
				.setParent(parent)
				.setReadSession(readSessionBuilder)
				.setMaxStreamCount(1)
				.build();

		// Init the read session
		final ReadSession readSession = readClient.createReadSession(createReadSessionRequest);
		if (readSession.getStreamsCount() == 0)
		{
			return Flux.fromIterable(new ArrayList<>());
		}

		// Get the streams
		final ReadStream readStream = readSession.getStreams(0);
		final ReadRowsRequest readRowsRequest = ReadRowsRequest.newBuilder()
				.setReadStream(readStream.getName())
				.build();
		final ServerStream<ReadRowsResponse> readRowsResponseStream = readClient
				.readRowsCallable()
				.call(readRowsRequest);

		// Allocate the limit
		final BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE);

		// Prepare the wrapper vectors
		final ArrowSchema arrowSchema = readSession.getArrowSchema();
		final Schema schema = MessageSerializer.deserializeSchema(
				new ReadChannel(new ByteArrayReadableSeekableByteChannel(arrowSchema.getSerializedSchema().toByteArray()))
		);
		final List<FieldVector> vectors = new ArrayList<>();
		for (Field field : schema.getFields())
		{
			vectors.add(field.createVector(allocator));
		}

		final VectorSchemaRoot vectorSchemaRoot = new VectorSchemaRoot(vectors);
		final VectorLoader vectorLoader = new VectorLoader(vectorSchemaRoot);

		final Iterable<ReadRowsResponse> iterable = new Iterable<ReadRowsResponse>()
		{
			Iterator<ReadRowsResponse> iterator = readRowsResponseStream.iterator();

			@NotNull
			@Override
			public Iterator<ReadRowsResponse> iterator()
			{
				return iterator;
			}
		};

		return Flux
				.fromIterable(iterable)
				.flatMapIterable(Unchecked.function(readRowsResponse -> {
					final ArrowRecordBatch batch = readRowsResponse.getArrowRecordBatch();
					final org.apache.arrow.vector.ipc.message.ArrowRecordBatch arrowRecordBatch = MessageSerializer
							.deserializeRecordBatch(
									new ReadChannel(
											new ByteArrayReadableSeekableByteChannel(
													batch.getSerializedRecordBatch().toByteArray())),
									allocator);
					vectorLoader.load(arrowRecordBatch);

					// Release buffers from batch (they are still held in the vectors in root).
					arrowRecordBatch.close();

					final List<FieldVector> sortedFieldVectors = vectorSchemaRoot.getFieldVectors()
							.stream()
							.sorted(Comparator.comparingLong((fv) -> api.getFieldNames().indexOf(fv.getName())))
							.collect(Collectors.toList());
					final int rowCount = vectorSchemaRoot.getRowCount();

					final ArrayList<T> records = new ArrayList<>(1024);
					final ArrayList<Object> row = new ArrayList<>(api.getFieldNames().size());
					for (int i = 0; i < rowCount; i++)
					{
						final ArrayList<Object> rowArrayList = api.getRowArrayList(row, sortedFieldVectors, i);
						final T record = api.fromList(rowArrayList);

						records.add(record);
					}

					// Release buffers from vectors in root.
					vectorSchemaRoot.clear();

					return records;
				}))
				.doOnCancel(() -> {
					logger.info("Cancel.");
				})
				.doFinally(Unchecked.consumer((signalType) -> {
					logger.info("signalType=[{}], iterator=[{}].", signalType, iterable.iterator());

					vectorSchemaRoot.close();
					allocator.close();

					readClient.shutdown();
					final boolean awaitTermination = readClient.awaitTermination(1, TimeUnit.MINUTES);
					logger.info(
							"signalType=[{}], iterator=[{}], awaitTermination=[{}].",
							signalType, iterable.iterator(), awaitTermination
					);
					readClient.close();
				}));
	}