Package com.netflix.genie.web.agent.apis.rpc.v4.endpoints

Class GRpcJobKillServiceImpl

java.lang.Object

com.netflix.genie.proto.JobKillServiceGrpc.JobKillServiceImplBase

com.netflix.genie.web.agent.apis.rpc.v4.endpoints.GRpcJobKillServiceImpl

All Implemented Interfaces:

JobKillService, io.grpc.BindableService

public class GRpcJobKillServiceImpl
extends com.netflix.genie.proto.JobKillServiceGrpc.JobKillServiceImplBase
implements JobKillService

Implementation of JobKillService which uses parked gRPC requests to tell the agent to shutdown via a user kill request if the job is in an active state.

Since:

4.0.0

Constructor Summary

Constructors

Constructor	Description
GRpcJobKillServiceImpl(DataServices dataServices, AgentRoutingService agentRoutingService, RequestForwardingService requestForwardingService)	Constructor.

Method Summary

Modifier and Type	Method	Description
void	cleanupOrphanedObservers()	Remove orphaned kill observers from local map.
protected boolean	isStreamObserverCancelled(io.grpc.stub.StreamObserver<com.netflix.genie.proto.JobKillRegistrationResponse> observer)	Converts StreamObserver into ServerCallStreamObserver in order to tell whether the observer is cancelled or not.
void	killJob(java.lang.String jobId, java.lang.String reason, javax.servlet.http.HttpServletRequest request)	Kill the job with the given id if possible.
void	registerForKillNotification(com.netflix.genie.proto.JobKillRegistrationRequest request, io.grpc.stub.StreamObserver<com.netflix.genie.proto.JobKillRegistrationResponse> responseObserver)	Register to be notified when a kill request for the job is received.

Methods inherited from class com.netflix.genie.proto.JobKillServiceGrpc.JobKillServiceImplBase

bindService

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

GRpcJobKillServiceImpl

public GRpcJobKillServiceImpl(DataServices dataServices,
                              AgentRoutingService agentRoutingService,
                              RequestForwardingService requestForwardingService)

Constructor.

Parameters:

dataServices - The DataServices instance to use

agentRoutingService - The AgentRoutingService instance to use to find where agents are connected

requestForwardingService - The service to use to forward requests to other Genie nodes

Method Detail

registerForKillNotification

public void registerForKillNotification(com.netflix.genie.proto.JobKillRegistrationRequest request,
                                        io.grpc.stub.StreamObserver<com.netflix.genie.proto.JobKillRegistrationResponse> responseObserver)

Register to be notified when a kill request for the job is received.

Overrides:

registerForKillNotification in class com.netflix.genie.proto.JobKillServiceGrpc.JobKillServiceImplBase

Parameters:

request - Request to register for getting notified when server gets a job kill request.

responseObserver - The response observer

killJob

@Retryable(value={GenieInvalidStatusException.class,GenieServerException.class},
           backoff=@Backoff(delay=1000L))
public void killJob(java.lang.String jobId,
                    java.lang.String reason,
                    @Nullable
                    javax.servlet.http.HttpServletRequest request)
             throws GenieJobNotFoundException,
                    GenieServerException

Kill the job with the given id if possible.

Specified by:

killJob in interface JobKillService

Parameters:

jobId - id of job to kill

reason - brief reason for requesting the job be killed

request - The optional HttpServletRequest information if the request needs to be forwarded

Throws:

GenieJobNotFoundException - When a job identified by jobId can't be found in the system

GenieServerException - if there is an unrecoverable error in the internal state of the Genie cluster

cleanupOrphanedObservers

@Scheduled(fixedDelay=30000L,
           initialDelay=30000L)
public void cleanupOrphanedObservers()

Remove orphaned kill observers from local map.

The logic as currently implemented is to have the Agent, once handshake is complete, open a connection to the server which results in parking a response observer in the map stored in this implementation. Upon receiving a kill request for the correct job this class will use the observer to send the "response" to the agent which will begin shut down process. The issue is that if the agent disconnects from this server the server will never realize it's gone and these observers will build up in the map in memory forever. This method will periodically go through the map and determine if the observers are still valid and remove any that aren't.

See Also:

"GRpcAgentJobKillServiceImpl"

isStreamObserverCancelled

protected boolean isStreamObserverCancelled(io.grpc.stub.StreamObserver<com.netflix.genie.proto.JobKillRegistrationResponse> observer)

Converts StreamObserver into ServerCallStreamObserver in order to tell whether the observer is cancelled or not.

Parameters:

observer - Observer for which we would check the status

Returns:

Boolean value: true if observer has status CANCELLED