Tempus 1053: Kinesis Support for Spark computations

application/src/main/java/com/hashmapinc/server/actors/computation/SparkComputationJobActorMessageProcessor.java

@@ -61,6 +61,7 @@ public class SparkComputationJobActorMessageProcessor extends ComponentMsgProces
     private ObjectMapper objectMapper = new ObjectMapper();
     private final ActorRef self;
     private final ActorRef parent;
+    private final String mainClass = "com.hashmapinc.tempus.Computation";
 
     protected SparkComputationJobActorMessageProcessor(TenantId tenantId, ComputationJobId id, ActorSystemContext systemContext
             , LoggingAdapter logger, ActorRef parent, ActorRef self, Computations computation) {
@@ -217,7 +218,7 @@ private String buildSparkComputationRequest() throws IOException {
         logger.info("Jar name is {}, main class is {}, arg parameters are {}, location is {}", md.getJarName(), md.getMainClass(), md.getArgsformat(), systemContext.getComputationLocation());
         SparkComputationRequest.SparkComputationRequestBuilder builder = SparkComputationRequest.builder();
         builder.file(systemContext.getComputationLocation() + md.getJarName());
-        builder.className(md.getMainClass());
+        builder.className(mainClass);
         builder.args(args());
         SparkComputationRequest sparkComputationRequest = builder.build();
         return objectMapper.writeValueAsString(sparkComputationRequest);
@@ -239,10 +240,40 @@ private String[] args() {
                 }
             }
         }
+        if(md.getArgsType().equals(ArgType.NAMED)){
+            args.add("--computation-class");
+            args.add(md.getMainClass());
+            args.add("--source");
+            String source = conf.get("source").asText();
+            args.add(source);
+            if("kinesis".equalsIgnoreCase(source)){
+                args.addAll(kinesisParameters(conf));
+            }else{
+                args.addAll(kafkaParameters(conf));
+            }
+        }
         logger.info("Argument array list to spark job " + args);
         return args.toArray(new String[args.size()]);
     }
 
+    private List<String> kinesisParameters(JsonNode conf){
+        List<String> kinesisArgs = new ArrayList<String>();
+        kinesisArgs.add("--kinesisStreamName");
+        kinesisArgs.add(conf.get("kinesisStreamName").asText());
+        kinesisArgs.add("--kinesisRegion");
+        kinesisArgs.add(conf.get("kinesisRegion").asText());
+        return kinesisArgs;
+    }
+
+    private List<String> kafkaParameters(JsonNode conf){
+        List<String> kafkaArgs = new ArrayList<String>();
+        kafkaArgs.add("--kafkaTopic");
+        kafkaArgs.add(conf.get("kafkaTopic").asText());
+        kafkaArgs.add("--kafkaUrl");
+        kafkaArgs.add(conf.get("kafkaUrl").asText());
+        return kafkaArgs;
+    }
+
     private void stopJobOnServer(){
         SparkComputationJob configuration = (SparkComputationJob) job.getConfiguration();
         if(configuration.getJobId() != null){

ui/src/app/components/json-form.directive.js

@@ -83,6 +83,7 @@ function JsonForm($compile, $templateCache, $mdColorPicker) {
                     val = undefined;
                 }
                 selectOrSet(key, scope.model, val);
+                scope.formProps.model = scope.model;
             },
             onColorClick: function(event, key, val) {
                 scope.showColorPicker(event, val);

ui/src/app/components/react/json-form-rc-select.jsx

@@ -92,7 +92,7 @@ class TempusRcSelect extends React.Component {
         }
 
         return (
-            <div className="tb-container">
+            <div className="tb-container" style={{'margin-top': '32px'}}>
                 <label className={labelClass}>{this.props.form.title}</label>
                 <Select
                     className={this.props.form.className}

ui/src/app/components/react/json-form-schema-form.jsx

@@ -64,11 +64,15 @@ class TempusSchemaForm extends React.Component {
 
         this.onChange = this.onChange.bind(this);
         this.onColorClick = this.onColorClick.bind(this);
+        this.hasConditions = false;
     }
 
     onChange(key, val) {
         //console.log('SchemaForm.onChange', key, val);
         this.props.onModelChange(key, val);
+        if (this.hasConditions) {
+            this.forceUpdate();
+        }
     }
 
     onColorClick(event, key, val) {
@@ -82,8 +86,12 @@ class TempusSchemaForm extends React.Component {
             console.log('Invalid field: \"' + form.key[0] + '\"!');
             return null;
         }
-        if(form.condition && eval(form.condition) === false) {
-            return null;
+
+        if(form.condition){
+            this.hasConditions = true;
+            if(eval(form.condition) === false) {
+                return null;
+            }
         }
 
         return <Field model={model} form={form} key={index} onChange={onChange} onColorClick={onColorClick} mapper={mapper} builder={this.builder}/>