Wie sollte ich einen Jetty-Server in einen Apache Ignite-Dienst einbetten?

Question

Ich versuche, einen Jetty-Server in einen Apache Ignite-Dienst einzubetten (gemäß this thread), damit ich den HTTP-Endpunkt zum Einstiegspunkt für meine Datenpipeline machen kann. Hier ist mein Basistest:

Main.scala

object Main { 
    def main(args: Array[String]): Unit = { 
     val ignite = Ignition.start() 
     val group = ignite.cluster.forLocal 
     val services = ignite.services(group) 
     services.deployNodeSingleton("myTestService", new TestServiceImpl) 
    } 
} 

TestService.scala

trait TestService { 
    def test() 
} 

class TestServiceImpl extends Service with TestService { 
    val server = new Server(8090) 

    def cancel(ctx: ServiceContext) = { 
     server.stop() 
     server.join() 
    } 

    def init(ctx: ServiceContext) = { 
     println("TestServiceImpl#init") 
    } 

    def execute(ctx: ServiceContext) = { 
     println("TestServiceImpl#execute") 
     server.start() 
    } 

    def test() = { 
     println("Tested") 
    } 
} 

Wenn ich es laufen lasse, erhalte ich folgende Fehlermeldung:

[01:52:57] Ignite node started OK (id=626c1302) 
[01:52:57] Topology snapshot [ver=1, servers=1, clients=0, CPUs=8, heap=2.0GB] 
TestServiceImpl#init 
TestServiceImpl#execute 
Oct 17, 2017 1:52:57 AM org.apache.ignite.logger.java.JavaLogger error 
SEVERE: Service execution stopped with error [name=myTestService, execId=565f4fb4-5726-4c37-857d-0c74f3b334ce] 
java.util.concurrent.RejectedExecutionException: org.eclipse.jetty.util.thread.NonBlockingThread@56a1831d 
    at org.eclipse.jetty.util.thread.QueuedThreadPool.execute(QueuedThreadPool.java:362) 
    at org.eclipse.jetty.io.SelectorManager.execute(SelectorManager.java:160) 
    at org.eclipse.jetty.io.SelectorManager.doStart(SelectorManager.java:258) 
    at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) 
    at org.eclipse.jetty.util.component.ContainerLifeCycle.start(ContainerLifeCycle.java:132) 
    at org.eclipse.jetty.util.component.ContainerLifeCycle.doStart(ContainerLifeCycle.java:106) 
    at org.eclipse.jetty.server.AbstractConnector.doStart(AbstractConnector.java:256) 
    at org.eclipse.jetty.server.AbstractNetworkConnector.doStart(AbstractNetworkConnector.java:81) 
    at org.eclipse.jetty.server.ServerConnector.doStart(ServerConnector.java:236) 
    at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) 
    at org.eclipse.jetty.server.Server.doStart(Server.java:366) 
    at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) 
    at me.danellis.ignite.TestServiceImpl.execute(TestService.scala:23) 
    at org.apache.ignite.internal.processors.service.GridServiceProcessor$2.run(GridServiceProcessor.java:1160) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
    at java.lang.Thread.run(Thread.java:748) 

Muss etwas sein in Ignite oder Jetty anders konfiguriert, damit dies funktioniert?

Answer 1

Sie sollten Anlegesteg Server-Klasse in init-Methode instanziiert, da es wird direkt nach der Bereitstellung des Dienstes auf dem Zielcomputer aufgerufen. Die Instanziierung der Server-Klasse im Konstruktor ist nutzlos - direkt nach dem Erstellen einer Service-Instanz (die in den meisten Fällen auf anderen Knoten ausgeführt werden kann) wird diese Instanz serialisiert und dem internen Cache hinzugefügt und erst nach dem Start auf dem Zielcomputer .

Ich denke, es ist klar, dass Anlegesteg Server-Objekt nicht ordnungsgemäß serialisiert werden kann. Zum Beispiel kann das Serialisieren von ThreadPool nicht durchgeführt werden, da die Thread-Implementierung Blöcke von nativem Code enthält.

Answer 2

Ich habe es gerade herausgefunden, aber ich werde es zugunsten anderer mit dem gleichen Problem verlassen.

Es stellt sich heraus, dass Sie die Server im Konstruktor nicht instanziieren können. Sie müssen tun es in der init Methode. (Keine Ahnung, warum, obwohl vielleicht die Thread-Pools sind noch nicht eingerichtet, wenn der Konstruktor ausgeführt wird.)

TestService.scala

class TestServiceImpl extends Service with TestService { 
    var server: Server = _ 

    def cancel(ctx: ServiceContext) = { 
     server.stop() 
     server.join() 
    } 

    def init(ctx: ServiceContext) = { 
     println("TestServiceImpl#init") 
     server = new Server(8090) 
    } 

    def execute(ctx: ServiceContext) = { 
     println("TestServiceImpl#execute") 
     server.start() 
    } 

    def test() = { 
     println("Tested") 
    } 
}