SAMZA-1852 Adding default job system in TestRunner, disabling host affinity to support TableDescriptors and refining addConfig method for TestRunner API #651

Sanil15 · 2018-09-20T23:49:40Z

The default system is a required config for intermediate streams, and since no user will write assertions against them, defaulting it makes it easier for the user to write test
To support stateful jobs using Table API descriptors we need to disable host affinity, which is enabled by table API by default
@vjagadish pointed out addConfigs vs addOverrideConfig to be a confusing user-facing API. We now support only addConfig with different signatures, this configs takes precedence over any descriptor or TestRunner generated configs

…pport TableDescriptors and refining addConfig method for TestRunner API

prateekm · 2018-09-21T18:13:45Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

    configs.put(JobCoordinatorConfig.JOB_COORDINATION_UTILS_FACTORY, PassthroughCoordinationUtilsFactory.class.getName());
    configs.put(JobCoordinatorConfig.JOB_COORDINATOR_FACTORY, PassthroughJobCoordinatorFactory.class.getName());
    configs.put(TaskConfig.GROUPER_FACTORY(), SingleContainerGrouperFactory.class.getName());
+    configs.put(JobConfig.JOB_DEFAULT_SYSTEM(), JOB_DEFAULT_SYSTEM);


Does this need to be addConfig as well? Users might be setting a default system descriptor in their application themselves.

Sure can add here

prateekm · 2018-09-21T18:14:19Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

+   * @param config for the application
+   * @return this {@link TestRunner}
+   */
+  private TestRunner putIfAbsentConfig(Map<String, String> config) {


Don't need a helper method for a one line method, can inline this.

descriptor.toConfig().forEach(configs::putIfAbsent); Java complains Incomptaible types cannot convert object to string

so two lines have to be added
Map<String, String> d = descriptor.toConfig();
d.forEach(configs::putIfAbsent)

Hence I made a method to avoid doing this all the time

@Sanil15 this is related to the type-erasure of generic types. If you declare descriptor as InMemoryInputDescriptor<?>, the one liner would work. Otherwise, since the descriptor is of an erased type InMemoryInputDescriptor, Java will complain about type conversion failure in compile time.

prateekm · 2018-09-21T18:15:05Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

  /**
-   * Only adds a config from {@code config} to samza job {@code configs} if they dont exist in it.
-   * @param config configs for the application
+   * Adds a config to Samza application, this config takes precedence over default configs and descriptor generated configs


Use period instead of comma for unrelated sentences. "Adds a ... application. This config ..."

prateekm · 2018-09-21T18:15:37Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

-  public TestRunner addOverrideConfig(String key, String value) {
-    Preconditions.checkNotNull(key);
-    Preconditions.checkNotNull(value);
+  public TestRunner addConfig(Config config) {


This should take a Map<String, String> instead of Config.

prateekm · 2018-09-21T18:16:11Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

+    Preconditions.checkNotNull(config);
    String configKeyPrefix = String.format(JobConfig.CONFIG_JOB_PREFIX(), JOB_NAME);
-    configs.put(String.format("%s%s", configKeyPrefix, key), value);
+    config.forEach((key, value) -> configs.put(String.format("%s%s", configKeyPrefix, key), value));


This can delegate to addConfig instead.

nickpan47

Thanks for putting up the patch. A few minor comments.

nickpan47 · 2018-09-24T20:02:51Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

+   * @param config for the application
+   * @return this {@link TestRunner}
+   */
+  private TestRunner putIfAbsentConfig(Map<String, String> config) {


@Sanil15 this is related to the type-erasure of generic types. If you declare descriptor as InMemoryInputDescriptor<?>, the one liner would work. Otherwise, since the descriptor is of an erased type InMemoryInputDescriptor, Java will complain about type conversion failure in compile time.

nickpan47 · 2018-09-24T20:13:50Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

        .createStream(spec);
-    addConfigs(streamDescriptor.toConfig());
-    addConfigs(streamDescriptor.getSystemDescriptor().toConfig());
+    putIfAbsentConfig(streamDescriptor.toConfig());


Can we add to the Javadoc of this method to explain why we are calling putIfAbsent here? I assume the reason is to allow user supplied config to supersede descriptor generated config? Can we make it clear?

nickpan47 · 2018-09-24T20:14:01Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

    imsd.withInMemoryScope(this.inMemoryScope);
-    addConfigs(descriptor.toConfig());
-    addConfigs(descriptor.getSystemDescriptor().toConfig());
+    putIfAbsentConfig(descriptor.toConfig());


nickpan47 · 2018-09-24T20:14:21Z

samza-test/src/test/java/org/apache/samza/test/framework/StreamTaskIntegrationTest.java

        .addInputStream(imid, inputList)
        .addOutputStream(imod, 1)
-        .addOverrideConfig("job.container.thread.pool.size", "4")
+        .addConfig("job.container.thread.pool.size", "4")


shanthoosh · 2018-09-22T07:14:03Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

 *
 */
 public class TestRunner {
+  public static final String JOB_DEFAULT_SYSTEM = "default-samza-system";


Set the visibility of both the static constants to private. These two constants are not used outside of TestRunner.

default job name is used in InMemorySystemDescriptor for jobs. prefix override

The usage of job name in InMemorySystemDescriptor is an anti-pattern that we should not encourage. I have already reverted that in PR #642 .

Reverted the anti-pattern, made the constants private

shanthoosh · 2018-09-24T16:33:37Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

    configs.put(TaskConfig.GROUPER_FACTORY(), SingleContainerGrouperFactory.class.getName());
+    addConfig(JobConfig.JOB_DEFAULT_SYSTEM(), JOB_DEFAULT_SYSTEM);
+    // This is important because Table Api enables host affinity by default for RocksDb
+    addConfig(ClusterManagerConfig.CLUSTER_MANAGER_HOST_AFFINITY_ENABLED, Boolean.FALSE.toString());


Shouldn't this be jobs.job_name.ClusterManagerConfig.CLUSTER_MANAGER_HOST_AFFINITY_ENABLED. Otherwise, it won't override the tableDescriptor generated configurations.

Please see the method addConfig(...), its adds that prefix

shanthoosh · 2018-09-24T20:03:17Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

  }
+
+  /**
+   * Only adds a config from {@code config} to samza job {@code configs} if they dont exist in it.


This java doc comment is confusing. Might be good to rename to the following(or something better):

Adds new configuration from {@param config} to samza application configuration {@code configs}

unnecessary now I removed it

shanthoosh · 2018-09-24T20:08:51Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

    imsd.withInMemoryScope(this.inMemoryScope);
-    addConfigs(descriptor.toConfig());
-    addConfigs(descriptor.getSystemDescriptor().toConfig());
+    putIfAbsentConfig(descriptor.toConfig());


putIfAbsentConfig API here is entirely unnecessary. You had to do this, since we didn't use the StreamMessageType as a type for InMemoryInputDescriptor in the API signature(rather used a raw type). Rather than the following(at line 369):

private <StreamMessageType> void initializeInMemoryInputStream(InMemoryInputDescriptor descriptor, Map<Integer, Iterable<StreamMessageType>> partitonData) { Map<Integer, Iterable<StreamMessageType>> partitonData) {

We can do the following :

private <StreamMessageType> void initializeInMemoryInputStream(InMemoryInputDescriptor<? extends StreeamMessageType> descriptor, Map<Integer, Iterable<StreamMessageType>> partitonData) { Map<Integer, Iterable<StreamMessageType>> partitonData) {

Since we haven't explicitly defined the type for genericly parameterized class, type-erasure kicks in and all the methods from InMemoryInputDescriptor return response of object type. If you do the above, then you don't have to do explicit typecast in all the places(For instance, at line 372 from Object to String).

After the above API signature change putifAbsentConfig API will be unnecessary, and could be simply in-lined it with the following:

descriptor.toConfig().forEach(this.configs::putIfAbsent);

If it does makes sense, then please change it.

nickpan47 · 2018-09-26T23:26:43Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

+  public TestRunner addConfig(String key, String value) {
+    Preconditions.checkNotNull(key);
+    Preconditions.checkNotNull(value);
+    String configKeyPrefix = String.format(JobConfig.CONFIG_JOB_PREFIX(), JOB_NAME);


Please refer to PR #642 , we need to have the full jobNameAndId in the config job prefix.

nickpan47 · 2018-09-26T23:28:46Z

samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java

        .createStream(spec);
-    addConfigs(streamDescriptor.toConfig());
-    addConfigs(streamDescriptor.getSystemDescriptor().toConfig());
+    streamDescriptor.toConfig().forEach(this.configs::putIfAbsent);


minor: add a comment here to explain why we use putIfAbsent, instead of overwriting.

Using the confix prefix, added a comment to docs

nickpan47

lgtm. Thanks!

Sanil15 added 2 commits September 20, 2018 16:33

Adding default job system in TestRunner, disabling hostaffinity to su…

5bbacdd

…pport TableDescriptors and refining addConfig method for TestRunner API

Correcting a typo

5738625

prateekm requested changes Sep 21, 2018

View reviewed changes

Sanil15 added 2 commits September 21, 2018 12:33

Addressing Review

0641242

Addressing a review

3c1167b

nickpan47 requested changes Sep 24, 2018

View reviewed changes

shanthoosh reviewed Sep 24, 2018

View reviewed changes

Adressing review

f44a472

nickpan47 reviewed Sep 26, 2018

View reviewed changes

Sanil15 added 2 commits September 28, 2018 12:30

Addressing review

b5b2931

Changing visibility of static variables

142a3f8

shanthoosh approved these changes Sep 28, 2018

View reviewed changes

Sanil15 added 2 commits September 28, 2018 15:20

Fixing merge conflicts

3fa83f8

Fixing LocalTable with side inputs test

f5c5821

nickpan47 approved these changes Oct 1, 2018

View reviewed changes

asfgit closed this in 094ff16 Oct 1, 2018

SAMZA-1852 Adding default job system in TestRunner, disabling host affinity to support TableDescriptors and refining addConfig method for TestRunner API #651

SAMZA-1852 Adding default job system in TestRunner, disabling host affinity to support TableDescriptors and refining addConfig method for TestRunner API #651

Uh oh!

Conversation

Sanil15 commented Sep 20, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Sanil15 Sep 21, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nickpan47 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shanthoosh Sep 24, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nickpan47 left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Sanil15 Sep 21, 2018 •

edited

Loading

shanthoosh Sep 24, 2018 •

edited

Loading