Merge remote-tracking branch 'upstream/master'

docs/_layouts/case-study.html

@@ -93,7 +93,7 @@ <h3>
 
           <div class="case-studies">
 
-              <div class="case-studies-list js-flickity" data-flickity-options='{ "wrapAround": true, "autoPlay": 1500 }'>
+              <div class="case-studies-list">
 
               {% assign case-studies_sorted = site.case-studies | sort: 'priority' %}
               {% for study in case-studies_sorted %}
@@ -124,8 +124,6 @@ <h3>
 
   {% include footer.html %}
 
-  <script type="text/javascript" src="/js/flickity.js"></script>
-
 </body>
 
 </html>

docs/_layouts/default.html

@@ -196,7 +196,7 @@ <h3>Latest from our blog</h3>
       Case Studies
     </div>
 
-    <div class="section__items js-flickity" data-flickity-options='{ "wrapAround": true, "autoPlay": 1500 }'>
+    <div class="section__items" >
     {% assign case-studies_sorted = site.case-studies | sort: 'priority' %}
     {% for study in case-studies_sorted %}
 
@@ -220,8 +220,6 @@ <h3>Latest from our blog</h3>
 
   {% include footer.html %}
 
-  <script type="text/javascript" src="/js/flickity.js"></script>
-
 </body>
 
 </html>

docs/css/main.new.css

@@ -750,72 +750,6 @@ footer .side-by-side > * {
   border: none;
 }
 
-.flickity-page-dots {
-  display: none;
-}
-
-.js-flickity {
-  position: relative;
-  display: block;
-  margin: auto;
-  width: 80%;
-  width: calc(100% - 100px);
-  margin-top: 40px;
-  outline: none;
-}
-
-.js-flickity::after {
-  content: '';
-  position: absolute;
-  top: 0;
-  right: 0;
-  width: 30%;
-  height: 100%;
-  background-image: linear-gradient(to right, rgba(255,255,255,0) 0%, transparent 20%, white 100%);
-  pointer-events: none;
-}
-
-.js-flickity::before {
-  content: '';
-  position: absolute;
-  top: 0;
-  left: 0;
-  width: 30%;
-  height: 100%;
-  background-image: linear-gradient(to left, rgba(255,255,255,0) 0%, transparent 20%, white 100%);
-  pointer-events: none;
-  z-index: 1;
-}
-
-.flickity-viewport {
-  overflow: hidden;
-}
-
-.flickity-button {
-  background: transparent;
-  position: absolute;
-  width: 50px;
-  height: 50px;
-  top: 50%;
-  border: none;
-  color: #333;
-  transform: translateY(-50%);
-  transition: fill .3s ease 0s;
-  cursor: pointer;
-}
-
-.flickity-button:hover {
-  fill: #eb1c23;
-}
-
-.flickity-button.previous {
-  left: -50px;
-}
-
-.flickity-button.next {
-  right: -50px;
-}
-
 /********************************************************************
  *
  * WEIRD MESH NODE CSS THING
@@ -900,7 +834,6 @@ footer .side-by-side > * {
 }
 
 
-
 .page .section {
   box-sizing: border-box;
   padding: 20px;
@@ -1539,10 +1472,6 @@ ul.case-studies {
   justify-content: center;
 }
 
-.case-studies-list.js-flickity {
-  display: block;
-}
-
 @media only screen and (min-width: 768px) {
   .case-studies-list {
     justify-content: start;

docs/learn/documentation/versioned/connectors/kafka.md

@@ -81,7 +81,7 @@ public class PageViewFilterTask implements TaskApplication {
     appDesc.addOutputStream(osd);
     appDesc.addTable(td);
 
-    appDesc.setTaskFactory((StreamTaskFactory) () -> new MyStreamTask());
+    appDesc.withTaskFactory((StreamTaskFactory) () -> new MyStreamTask());
   }
 }
 {% endhighlight %}

docs/learn/documentation/versioned/core-concepts/core-concepts.md

@@ -1,4 +1,4 @@
-.png
+---
 layout: page
 title: Core concepts
 ---

samza-api/src/main/java/org/apache/samza/application/TaskApplication.java

@@ -36,7 +36,7 @@
  *   {@link org.apache.samza.task.AsyncStreamTask} that operates on each
  *   {@link org.apache.samza.system.IncomingMessageEnvelope} one at a time.
  *   <li>Setting a {@link org.apache.samza.task.TaskFactory} using
- *   {@link TaskApplicationDescriptor#setTaskFactory(org.apache.samza.task.TaskFactory)} that creates instances of the
+ *   {@link TaskApplicationDescriptor#withTaskFactory(org.apache.samza.task.TaskFactory)} that creates instances of the
  *   task above. The {@link org.apache.samza.task.TaskFactory} implementation must be {@link java.io.Serializable}.
  * </ol>
  * <p>
@@ -50,9 +50,10 @@
  *     KafkaOutputDescriptor<PageViewEvent>> outputStreamDescriptor =
  *         trackingSystemDescriptor.getOutputDescriptor("recentPageViewEvent", new JsonSerdeV2<>(PageViewEvent.class)));
  *
- *     appDescriptor.addInputStream(inputStreamDescriptor);
- *     appDescriptor.addOutputStream(outputStreamDescriptor);
- *     appDescriptor.setTaskFactory((StreamTaskFactory) () -> new PageViewTask());
+ *     appDescriptor
+ *         .withInputStream(inputStreamDescriptor)
+ *         .withOutputStream(outputStreamDescriptor)
+ *         .withTaskFactory((StreamTaskFactory) () -> new PageViewTask());
  *   }
  * }
  *

samza-api/src/main/java/org/apache/samza/application/descriptors/TaskApplicationDescriptor.java

@@ -33,7 +33,7 @@
  * to add the {@link InputDescriptor}s, {@link OutputDescriptor}s and {@link TableDescriptor}s for streams and
  * tables to be used in the task implementation.
  * <p>
- * Use {@link #setTaskFactory} to set the factory for the {@link org.apache.samza.task.StreamTask} or
+ * Use {@link #withTaskFactory} to set the factory for the {@link org.apache.samza.task.StreamTask} or
  * {@link org.apache.samza.task.AsyncStreamTask} implementation that contains the processing logic for
  * the {@link org.apache.samza.application.TaskApplication}.
  */
@@ -48,28 +48,32 @@ public interface TaskApplicationDescriptor extends ApplicationDescriptor<TaskApp
    * The provided {@code taskFactory} instance must be serializable.
    *
    * @param factory the {@link TaskFactory} for the Low Level API Task implementation
+   * @return this {@link TaskApplicationDescriptor}
    */
-  void setTaskFactory(TaskFactory factory);
+  TaskApplicationDescriptor withTaskFactory(TaskFactory factory);
 
   /**
    * Adds the input stream to the application.
    *
    * @param isd the {@link InputDescriptor}
+   * @return this {@link TaskApplicationDescriptor}
    */
-  void addInputStream(InputDescriptor isd);
+  TaskApplicationDescriptor withInputStream(InputDescriptor isd);
 
   /**
    * Adds the output stream to the application.
    *
    * @param osd the {@link OutputDescriptor} of the output stream
+   * @return this {@link TaskApplicationDescriptor}
    */
-  void addOutputStream(OutputDescriptor osd);
+  TaskApplicationDescriptor withOutputStream(OutputDescriptor osd);
 
   /**
    * Adds the {@link TableDescriptor} used in the application
    *
    * @param table {@link TableDescriptor}
+   * @return this {@link TaskApplicationDescriptor}
    */
-  void addTable(TableDescriptor table);
+  TaskApplicationDescriptor withTable(TableDescriptor table);
 
 }

samza-core/src/main/java/org/apache/samza/application/LegacyTaskApplication.java

@@ -33,6 +33,6 @@ public LegacyTaskApplication(String taskClassName) {
 
   @Override
   public void describe(TaskApplicationDescriptor appDescriptor) {
-    appDescriptor.setTaskFactory(TaskFactoryUtil.getTaskFactory(taskClassName));
+    appDescriptor.withTaskFactory(TaskFactoryUtil.getTaskFactory(taskClassName));
   }
 }

samza-core/src/main/java/org/apache/samza/application/descriptors/TaskApplicationDescriptorImpl.java

@@ -44,28 +44,32 @@ public TaskApplicationDescriptorImpl(TaskApplication userApp, Config config) {
   }
 
   @Override
-  public void setTaskFactory(TaskFactory factory) {
+  public TaskApplicationDescriptor withTaskFactory(TaskFactory factory) {
     this.taskFactory = factory;
+    return this;
   }
 
   @Override
-  public void addInputStream(InputDescriptor inputDescriptor) {
+  public TaskApplicationDescriptor withInputStream(InputDescriptor inputDescriptor) {
     // TODO: SAMZA-1841: need to add to the broadcast streams if inputDescriptor is for a broadcast stream
     addInputDescriptor(inputDescriptor);
     getOrCreateStreamSerdes(inputDescriptor.getStreamId(), inputDescriptor.getSerde());
+    return this;
   }
 
   @Override
-  public void addOutputStream(OutputDescriptor outputDescriptor) {
+  public TaskApplicationDescriptor withOutputStream(OutputDescriptor outputDescriptor) {
     addOutputDescriptor(outputDescriptor);
     getOrCreateStreamSerdes(outputDescriptor.getStreamId(), outputDescriptor.getSerde());
+    return this;
   }
 
   @Override
-  public void addTable(TableDescriptor tableDescriptor) {
+  public TaskApplicationDescriptor withTable(TableDescriptor tableDescriptor) {
     addTableDescriptor(tableDescriptor);
     BaseTableDescriptor baseTableDescriptor = (BaseTableDescriptor) tableDescriptor;
     getOrCreateTableSerdes(baseTableDescriptor.getTableId(), baseTableDescriptor.getSerde());
+    return this;
   }
 
   /**

samza-core/src/main/java/org/apache/samza/execution/JobPlanner.java

@@ -70,17 +70,23 @@ ExecutionPlan getExecutionPlan() {
 
   /* package private */
   ExecutionPlan getExecutionPlan(String runId) {
-    Map<String, String> generatedConfig = getGeneratedConfig(runId);
+    Map<String, String> allowedUserConfig = new HashMap<>(userConfig);
+    Map<String, String> generatedConfig = new HashMap<>();
 
-    // merge user-provided configuration with generated configuration. generated configuration has lower priority.
     // TODO: This should all be consolidated with ExecutionPlanner after fixing SAMZA-1811
-    Map<String, String> allowedUserConfig = new HashMap<>(userConfig);
+    // Don't generate any configurations for LegacyTaskApplications
     if (!LegacyTaskApplication.class.isAssignableFrom(appDesc.getAppClass())) {
-      LOG.warn("SamzaApplications should not specify task.inputs in configuration. " +
-          "Ignoring configured value of " + userConfig.get(TaskConfig.INPUT_STREAMS()));
-      allowedUserConfig.remove(TaskConfig.INPUT_STREAMS()); // must be set using descriptors or operators
+      if (userConfig.containsKey(TaskConfig.INPUT_STREAMS())) {
+        LOG.warn("SamzaApplications should not specify task.inputs in configuration. " +
+            "Specify them using InputDescriptors instead. Ignoring configured task.inputs value of " +
+            userConfig.get(TaskConfig.INPUT_STREAMS()));
+        allowedUserConfig.remove(TaskConfig.INPUT_STREAMS());
+      }
+
+      generatedConfig.putAll(getGeneratedConfig(runId));
     }
 
+    // merge user-provided configuration with generated configuration. generated configuration has lower priority.
     Config mergedConfig = JobNodeConfigurationGenerator.mergeConfig(allowedUserConfig, generatedConfig);
 
     // creating the StreamManager to get all input/output streams' metadata for planning