I have single node data torrent/Apex installation. Software installed under
"/opt/datatorrent/releases/3.2.0”. When I uploaded and launched my application
I get error finding the class “KafkaSinglePortStringInputOperator”. I could
compile the application after adding the dt-contrib dependency but not sure how
dt-contrib jars be available for application when running on single node data
torrent installation.
Appreciate help!
—Suhas
PS: I explicitly copied the dt-contrib-3.1.1.jar to
/opt/datatorrent/current/lib/dt-contrib-3.1.1.jar and restarted the gateway..
Error launching the application:
An error occurred trying to launch the application. Server message:
java.lang.NoClassDefFoundError:
com/datatorrent/contrib/kafka/KafkaSinglePortStringInputOperator at
io.ampool.demo.adtech.Application.populateDAG(Application.java:44) at
com.datatorrent.stram.plan.logical.LogicalPlanConfiguration.prepareDAG(LogicalPlanConfiguration.java:2108)
at
com.datatorrent.stram.client.StramAppLauncher$1.createApp(StramAppLauncher.java:407)
at
com.datatorrent.stram.client.StramAppLauncher.launchApp(StramAppLauncher.java:482)
at com.datatorrent.stram.cli.DTCli$LaunchCommand.execute(DTCli.java:2047) at
com.datatorrent.stram.cli.DTCli.launchAppPackage(DTCli.java:3450) at
com.datatorrent.stram.cli.DTCli.access$7000(DTCli.java:106) at
com.datatorrent.stram.cli.DTCli$LaunchCommand.execute(DTCli.java:1892) at
com.datatorrent.stram.cli.DTCli$3.run(DTCli.java:1449) Caused by:
java.lang.ClassNotFoundException:
com.datatorrent.contrib.kafka.KafkaSinglePortStringInputOperator at
java.net.URLClassLoader.findClass(URLClassLoader.java:381) at
java.lang.ClassLoader.loadClass(ClassLoader.java:424) at
java.net.FactoryURLClassLoader.loadClass(URLClassLoader.java:810) at
java.lang.ClassLoader.loadClass(ClassLoader.java:357) ... 9 more Fatal error
encountered
Application.java
import com.datatorrent.api.StreamingApplication;
import com.datatorrent.api.DAG;
import com.datatorrent.api.DAG.Locality;
import com.datatorrent.lib.io.ConsoleOutputOperator;
import com.datatorrent.contrib.kafka.KafkaSinglePortStringInputOperator;
@ApplicationAnnotation(name = "AdStream")
public class Application implements StreamingApplication
{
@Override
public void populateDAG(DAG dag, Configuration entries)
{
KafkaSinglePortStringInputOperator input =
dag.addOperator("MessageReader", new KafkaSinglePortStringInputOperator());
ConsoleOutputOperator output = dag.addOperator("Output", new
ConsoleOutputOperator());
dag.addStream("MessageData", input.outputPort, output.input);
}
}
