private void fetchAndParseImpl(FetchAndParseRequest request,
                                   StreamObserver<FetchAndParseReply> responseObserver) {
        AbstractFetcher fetcher =
                expiringFetcherStore.getFetcherAndLogAccess(request.getFetcherId());
        if (fetcher == null) {
            throw new RuntimeException(
                    "Could not find fetcher with name " + request.getFetcherId());
        }
        Metadata tikaMetadata = new Metadata();
        try {
            ParseContext parseContext = new ParseContext();
            String additionalFetchConfigJson = request.getAdditionalFetchConfigJson();
            if (StringUtils.isNotBlank(additionalFetchConfigJson)) {
                // The fetch and parse has the option to specify additional configuration
                AbstractConfig abstractConfig = expiringFetcherStore
                        .getFetcherConfigs()
                        .get(fetcher.getName());
                parseContext.set(FetcherConfigContainer.class, new FetcherConfigContainer()
                        .setConfigClassName(abstractConfig
                                .getClass().getName())
                        .setJson(additionalFetchConfigJson));
            }
            PipesResult pipesResult = pipesClient.process(new FetchEmitTuple(request.getFetchKey(),
                    new FetchKey(fetcher.getName(), request.getFetchKey()), new EmitKey(), tikaMetadata, parseContext, FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
            FetchAndParseReply.Builder fetchReplyBuilder =
                    FetchAndParseReply.newBuilder()
                                      .setFetchKey(request.getFetchKey())
                            .setStatus(pipesResult.getStatus().name());
            if (pipesResult.getStatus().equals(PipesResult.STATUS.FETCH_EXCEPTION)) {
                fetchReplyBuilder.setErrorMessage(pipesResult.getMessage());
            }
            if (pipesResult.getEmitData() != null && pipesResult.getEmitData().getMetadataList() != null) {
                for (Metadata metadata : pipesResult.getEmitData().getMetadataList()) {
                    for (String name : metadata.names()) {
                        String value = metadata.get(name);
                        if (value != null) {
                            fetchReplyBuilder.putFields(name, value);
                        }
                    }
                }
            }
            responseObserver.onNext(fetchReplyBuilder.build());
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
FetcherID is key, The pair is the Fetcher object and the Metadata
tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java