core/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java
protected int sendSnapshot(final Translog.Snapshot snapshot) {
        int ops = 0;
        long size = 0;
        int totalOperations = 0;
        final List<Translog.Operation> operations = new ArrayList<>();
        Translog.Operation operation;
        try {
            operation = snapshot.next(); // this ex should bubble up
        } catch (IOException ex) {
            throw new ElasticsearchException("failed to get next operation from translog", ex);
        }

        final TransportRequestOptions recoveryOptions = TransportRequestOptions.builder()
                .withCompress(true)
                .withType(TransportRequestOptions.Type.RECOVERY)
                .withTimeout(recoverySettings.internalActionLongTimeout())
                .build();

        if (operation == null) {
            logger.trace("[{}][{}] no translog operations to send to {}",
                    indexName, shardId, request.targetNode());
        }
        while (operation != null) {
            if (shard.state() == IndexShardState.CLOSED) {
                throw new IndexShardClosedException(request.shardId());
            }
            cancellableThreads.checkForCancel();
            operations.add(operation);
            ops += 1;
            size += operation.estimateSize();
            totalOperations++;

            // Check if this request is past bytes threshold, and
            // if so, send it off
            if (size >= chunkSizeInBytes) {

                // don't throttle translog, since we lock for phase3 indexing,
                // so we need to move it as fast as possible. Note, since we
                // index docs to replicas while the index files are recovered
                // the lock can potentially be removed, in which case, it might
                // make sense to re-enable throttling in this phase
                cancellableThreads.execute(() -> {
                    final RecoveryTranslogOperationsRequest translogOperationsRequest = new RecoveryTranslogOperationsRequest(
                            request.recoveryId(), request.shardId(), operations, snapshot.totalOperations());
                    transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.TRANSLOG_OPS, translogOperationsRequest,
                            recoveryOptions, EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
                });
                if (logger.isTraceEnabled()) {
                    logger.trace("[{}][{}] sent batch of [{}][{}] (total: [{}]) translog operations to {}",
                            indexName, shardId, ops, new ByteSizeValue(size),
                            snapshot.totalOperations(),
                            request.targetNode());
                }

                ops = 0;
                size = 0;
                operations.clear();
            }
            try {
                operation = snapshot.next(); // this ex should bubble up
            } catch (IOException ex) {
                throw new ElasticsearchException("failed to get next operation from translog", ex);
            }
        }
        // send the leftover
        if (!operations.isEmpty()) {
            cancellableThreads.execute(() -> {
                RecoveryTranslogOperationsRequest translogOperationsRequest = new RecoveryTranslogOperationsRequest(
                        request.recoveryId(), request.shardId(), operations, snapshot.totalOperations());
                transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.TRANSLOG_OPS, translogOperationsRequest,
                        recoveryOptions, EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
            });

        }
        if (logger.isTraceEnabled()) {
            logger.trace("[{}][{}] sent final batch of [{}][{}] (total: [{}]) translog operations to {}",
                    indexName, shardId, ops, new ByteSizeValue(size),
                    snapshot.totalOperations(),
                    request.targetNode());
        }
        return totalOperations;
    }
/**
     * Send the given snapshot's operations to this handler's target node.
     * <p>
     * Operations are bulked into a single request depending on an operation
     * count limit or size-in-bytes limit
     *
     * @return the total number of translog operations that were sent
     */
Send the given snapshot's operations to this handler's target node.

Operations are bulked into a single request depending on an operation
count limit or size-in-bytes limit

@return the total number of translog operations that were sent
Send the given snapshot ' s operations to this handler ' s target node . Operations are bulked into a single request depending on an operation count limit or size - in - bytes limit @return the total number of translog operations that were sent
protected int sendSnapshot(final Translog.Snapshot snapshot) {
        int ops = 0;
        long size = 0;
        int totalOperations = 0;
        final List<Translog.Operation> operations = new ArrayList<>();
        Translog.Operation operation;
        try {
            operation = snapshot.next(); 
        } catch (IOException ex) {
            throw new ElasticsearchException("failed to get next operation from translog", ex);
        }
        final TransportRequestOptions recoveryOptions = TransportRequestOptions.builder()
                .withCompress(true)
                .withType(TransportRequestOptions.Type.RECOVERY)
                .withTimeout(recoverySettings.internalActionLongTimeout())
                .build();
        if (operation == null) {
            logger.trace("[{}][{}] no translog operations to send to {}",
                    indexName, shardId, request.targetNode());
        }
        while (operation != null) {
            if (shard.state() == IndexShardState.CLOSED) {
                throw new IndexShardClosedException(request.shardId());
            }
            cancellableThreads.checkForCancel();
            operations.add(operation);
            ops += 1;
            size += operation.estimateSize();
            totalOperations++;
            if (size >= chunkSizeInBytes) {
                cancellableThreads.execute(() -> {
                    final RecoveryTranslogOperationsRequest translogOperationsRequest = new RecoveryTranslogOperationsRequest(
                            request.recoveryId(), request.shardId(), operations, snapshot.totalOperations());
                    transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.TRANSLOG_OPS, translogOperationsRequest,
                            recoveryOptions, EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
                });
                if (logger.isTraceEnabled()) {
                    logger.trace("[{}][{}] sent batch of [{}][{}] (total: [{}]) translog operations to {}",
                            indexName, shardId, ops, new ByteSizeValue(size),
                            snapshot.totalOperations(),
                            request.targetNode());
                }
                ops = 0;
                size = 0;
                operations.clear();
            }
            try {
                operation = snapshot.next(); 
            } catch (IOException ex) {
                throw new ElasticsearchException("failed to get next operation from translog", ex);
            }
        }
        if (!operations.isEmpty()) {
            cancellableThreads.execute(() -> {
                RecoveryTranslogOperationsRequest translogOperationsRequest = new RecoveryTranslogOperationsRequest(
                        request.recoveryId(), request.shardId(), operations, snapshot.totalOperations());
                transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.TRANSLOG_OPS, translogOperationsRequest,
                        recoveryOptions, EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
            });
        }
        if (logger.isTraceEnabled()) {
            logger.trace("[{}][{}] sent final batch of [{}][{}] (total: [{}]) translog operations to {}",
                    indexName, shardId, ops, new ByteSizeValue(size),
                    snapshot.totalOperations(),
                    request.targetNode());
        }
        return totalOperations;
    }
protected int sendSnapshot ( final Translog . Snapshot snapshot ) { int ops = 0 ; long size = 0 ; int totalOperations = 0 ; final List < Translog . Operation > operations = new ArrayList < > ( ) ; Translog . Operation operation ; try { operation = snapshot . next ( ) ; } catch ( IOException ex ) { throw new ElasticsearchException ( "failed to get next operation from translog" , ex ) ; } final TransportRequestOptions recoveryOptions = TransportRequestOptions . builder ( ) . withCompress ( true ) . withType ( TransportRequestOptions . Type . RECOVERY ) . withTimeout ( recoverySettings . internalActionLongTimeout ( ) ) . build ( ) ; if ( operation == null ) { logger . trace ( "[{}][{}] no translog operations to send to {}" , indexName , shardId , request . targetNode ( ) ) ; } while ( operation != null ) { if ( shard . state ( ) == IndexShardState . CLOSED ) { throw new IndexShardClosedException ( request . shardId ( ) ) ; } cancellableThreads . checkForCancel ( ) ; operations . add ( operation ) ; ops += 1 ; size += operation . estimateSize ( ) ; totalOperations ++ ; if ( size >= chunkSizeInBytes ) { cancellableThreads . execute ( ( ) -> { final RecoveryTranslogOperationsRequest translogOperationsRequest = new RecoveryTranslogOperationsRequest ( request . recoveryId ( ) , request . shardId ( ) , operations , snapshot . totalOperations ( ) ) ; transportService . submitRequest ( request . targetNode ( ) , RecoveryTarget . Actions . TRANSLOG_OPS , translogOperationsRequest , recoveryOptions , EmptyTransportResponseHandler . INSTANCE_SAME ) . txGet ( ) ; } ) ; if ( logger . isTraceEnabled ( ) ) { logger . trace ( "[{}][{}] sent batch of [{}][{}] (total: [{}]) translog operations to {}" , indexName , shardId , ops , new ByteSizeValue ( size ) , snapshot . totalOperations ( ) , request . targetNode ( ) ) ; } ops = 0 ; size = 0 ; operations . clear ( ) ; } try { operation = snapshot . next ( ) ; } catch ( IOException ex ) { throw new ElasticsearchException ( "failed to get next operation from translog" , ex ) ; } } if ( ! operations . isEmpty ( ) ) { cancellableThreads . execute ( ( ) -> { RecoveryTranslogOperationsRequest translogOperationsRequest = new RecoveryTranslogOperationsRequest ( request . recoveryId ( ) , request . shardId ( ) , operations , snapshot . totalOperations ( ) ) ; transportService . submitRequest ( request . targetNode ( ) , RecoveryTarget . Actions . TRANSLOG_OPS , translogOperationsRequest , recoveryOptions , EmptyTransportResponseHandler . INSTANCE_SAME ) . txGet ( ) ; } ) ; } if ( logger . isTraceEnabled ( ) ) { logger . trace ( "[{}][{}] sent final batch of [{}][{}] (total: [{}]) translog operations to {}" , indexName , shardId , ops , new ByteSizeValue ( size ) , snapshot . totalOperations ( ) , request . targetNode ( ) ) ; } return totalOperations ; }
Send the given snapshot's operations to this handler's target node.
Send the given snapshot ' s operations to this handler ' s target node .
// this ex should bubble up // Check if this request is past bytes threshold, and // if so, send it off // don't throttle translog, since we lock for phase3 indexing, // so we need to move it as fast as possible. Note, since we // index docs to replicas while the index files are recovered // the lock can potentially be removed, in which case, it might // make sense to re-enable throttling in this phase // this ex should bubble up // send the leftover