diff --git a/build.gradle b/build.gradle
index a3512e0..2b98b8a 100644
--- a/build.gradle
+++ b/build.gradle
@@ -127,11 +127,6 @@
     toolVersion = "$checkstyleVersion"
   }
 
-  tasks.withType(ScalaCompile) {
-    // show compile errors in console output
-    logging.setLevel LogLevel.WARN
-  }
-
   tasks.withType(Test) {
     test {
       testLogging {
@@ -196,7 +191,6 @@
     compile "org.scala-lang:scala-library:$scalaVersion"
     compile "org.slf4j:slf4j-api:$slf4jVersion"
     compile "net.jodah:failsafe:$failsafeVersion"
-    compile "com.linkedin.cytodynamics:cytodynamics-nucleus:$cytodynamicsVersion"
     testCompile project(":samza-api").sourceSets.test.output
     testCompile "junit:junit:$junitVersion"
     testCompile "org.mockito:mockito-core:$mockitoVersion"
@@ -221,6 +215,7 @@
 
   dependencies {
     compile "com.azure:azure-storage-blob:12.0.1"
+    compile "com.azure:azure-identity:1.0.0"
     compile "com.microsoft.azure:azure-storage:5.3.1"
     compile "com.microsoft.azure:azure-eventhubs:1.0.1"
     compile "com.fasterxml.jackson.core:jackson-core:$jacksonVersion"
diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html
index b588e64..18a782f 100644
--- a/docs/learn/documentation/versioned/jobs/configuration-table.html
+++ b/docs/learn/documentation/versioned/jobs/configuration-table.html
@@ -1912,6 +1912,22 @@
                 </tr>
 
                 <tr>
+                    <td class="property" id="stores-rocksdb-max-open-files">stores.<span class="store">store-name</span>.<br>rocksdb.max.open.files</td>
+                    <td class="default">-1</td>
+                    <td class="description">
+                        Limits the number of open files that RocksDB can have open at one time.
+                    </td>
+                </tr>
+
+                <tr>
+                    <td class="property" id="stores-rocksdb-max-file-opening-threads">stores.<span class="store">store-name</span>.<br>rocksdb.max.file.opening.threads</td>
+                    <td class="default">16</td>
+                    <td class="description">
+                        Sets the number of threads used to open RocksDB files.
+                    </td>
+                </tr>
+
+                <tr>
                     <td class="property" id="stores-rocksdb-metrics">stores.<span class="store">store-name</span>.<br>rocksdb.metrics.list</td>
                     <td class="default"></td>
                     <td class="description">
diff --git a/docs/learn/documentation/versioned/jobs/samza-configurations.md b/docs/learn/documentation/versioned/jobs/samza-configurations.md
index 596dc52..1ef74cd 100644
--- a/docs/learn/documentation/versioned/jobs/samza-configurations.md
+++ b/docs/learn/documentation/versioned/jobs/samza-configurations.md
@@ -258,7 +258,7 @@
 #### <a name="advanced-azure-blob-storage"></a>[Advanced Azure Blob Storage Configurations](#advanced-azure-blob-storage)
 |Name|Default|Description|
 |--- |--- |--- |
-|systems.**_system-name_**.azureblob.proxy.use |false|if true, proxy will be used to connect to Azure.|
+|systems.**_system-name_**.azureblob.proxy.use |false|if true, proxy will be used to connect to Azure for blob creation.|
 |systems.**_system-name_**.azureblob.proxy.hostname| |if proxy.use is true then host name of proxy.|
 |systems.**_system-name_**.azureblob.proxy.port| |if proxy.use is true then port of proxy.|
 |systems.**_system-name_**.azureblob.writer.factory.class|`org.apache.samza.system.`<br>`azureblob.avro.`<br>`AzureBlobAvroWriterFactory`|Fully qualified class name of the `org.apache.samza.system.azureblob.producer.AzureBlobWriter` impl for the system producer.<br><br>The default writer creates blobs that are of type AVRO and require the messages sent to a blob to be AVRO records. The blobs created by the default writer are of type [Block Blobs](https://docs.microsoft.com/en-us/rest/api/storageservices/understanding-block-blobs--append-blobs--and-page-blobs#about-block-blobs).|
@@ -272,7 +272,14 @@
 |systems.**_system-name_**.azureblob.closeTimeoutMs|300000 (5 mins)|timeout to finish committing all the blobs currently being written to. This does not include the flush timeout per blob.|
 |systems.**_system-name_**.azureblob.suffixRandomStringToBlobName|true|if true, a random string of 8 chars is suffixed to the blob name to prevent name collision when more than one Samza tasks are writing to the same SSP.|
 |systems.**_system-name_**.azureblob.metadataPropertiesGeneratorFactory|`org.apache.samza.system.`<br>`azureblob.utils.`<br>`NullBlobMetadataGeneratorFactory`|Fully qualified class name of the `org.apache.samza.system.azureblob.utils.BlobMetadataGeneratorFactory` impl for the system producer. <br><br>The default metadata generator does not add any metadata to the blob.| 
-|systems.**_system-name_**.azureblob.metadataGeneratorConfig| |Additional configs for the metadata generator should be prefixed with this string which is passed to the generator.<br>For example, to pass a "key":"value" pair to the metadata generator, add config like systems.<system-name>.azureblob.metadataGeneratorConfig.\<key\> with value \<value\>| 
+|systems.**_system-name_**.azureblob.metadataGeneratorConfig| |Additional configs for the metadata generator should be prefixed with this string which is passed to the generator.<br>For example, to pass a "key":"value" pair to the metadata generator, add config like systems.<system-name>.azureblob.metadataGeneratorConfig.\<key\> with value \<value\>|
+|systems.**_system-name_**.azureblob.useTokenCredentialAuthentication|false| if true, then com.azure.core.credential.TokenCredential is used to authenticate with Azure.|
+|systems.**_system-name_**.azureblob.client.id| | if TokenCredential authentication, then the client id to be used for authentication.|
+|systems.**_system-name_**.azureblob.client.secret| | if TokenCredential authentication, then the client secret to be used for authentication.|
+|systems.**_system-name_**.azureblob.tenant.id| | if TokenCredential authentication, then the tenant id to be used for authentication.|
+|systems.**_system-name_**.azureblob.authProxy.use|false| if TokenCredential authentication, then setting this to true will result in using a proxy for authentication.|
+|systems.**_system-name_**.azureblob.authProxy.hostName| |if authProxy.use is true then host name of proxy.|
+|systems.**_system-name_**.azureblob.authProxy.port| |if authProxy.use is true then port of proxy.|
 
 
 ### <a name="state-storage"></a>[4. State Storage](#state-storage)
diff --git a/gradle.properties b/gradle.properties
index d7c4cd7..1a6ee97 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -25,3 +25,5 @@
 
 systemProp.file.encoding=utf-8
 checkstyleVersion=6.11.2
+
+org.gradle.parallel=true
diff --git a/gradle/dependency-versions.gradle b/gradle/dependency-versions.gradle
index ebd2d38..019cde3 100644
--- a/gradle/dependency-versions.gradle
+++ b/gradle/dependency-versions.gradle
@@ -25,7 +25,6 @@
   commonsHttpClientVersion = "3.1"
   commonsIoVersion = "2.8.0"
   commonsLang3Version = "3.11"
-  cytodynamicsVersion = "0.2.0"
   elasticsearchVersion = "2.2.0"
   gsonVersion = "2.8.6"
   guavaVersion = "30.1-jre"
diff --git a/samza-azure/src/main/java/org/apache/samza/system/azureblob/AzureBlobClientBuilder.java b/samza-azure/src/main/java/org/apache/samza/system/azureblob/AzureBlobClientBuilder.java
new file mode 100644
index 0000000..b036dc6
--- /dev/null
+++ b/samza-azure/src/main/java/org/apache/samza/system/azureblob/AzureBlobClientBuilder.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.system.azureblob;
+
+import com.azure.core.credential.TokenCredential;
+import com.azure.core.http.ProxyOptions;
+import com.azure.core.http.netty.NettyAsyncHttpClientBuilder;
+import com.azure.core.http.policy.HttpLogDetailLevel;
+import com.azure.core.http.policy.HttpLogOptions;
+import com.azure.identity.ClientSecretCredentialBuilder;
+import com.azure.storage.blob.BlobServiceAsyncClient;
+import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.common.StorageSharedKeyCredential;
+import java.net.InetSocketAddress;
+import com.azure.core.http.HttpClient;
+import java.util.Locale;
+
+
+/**
+ * This class provides a method to create {@link BlobServiceAsyncClient} to be used by the
+ * {@link org.apache.samza.system.azureblob.producer.AzureBlobSystemProducer}. It created the client based on the
+ * configs given to the SystemProducer - such as which authentication method to use, whether to use proxy to authenticate,
+ * and so on.
+ */
+public final class AzureBlobClientBuilder {
+  private final String systemName;
+  private final String azureUrlFormat;
+  private final AzureBlobConfig azureBlobConfig;
+  public AzureBlobClientBuilder(String systemName, String azureUrlFormat, AzureBlobConfig azureBlobConfig) {
+    this.systemName = systemName;
+    this.azureUrlFormat = azureUrlFormat;
+    this.azureBlobConfig = azureBlobConfig;
+  }
+
+  /**
+   * method creates BlobServiceAsyncClient using the configs provided earlier.
+   * if the authentication method is set to {@link TokenCredential} then a {@link com.azure.identity.ClientSecretCredential}
+   * is created and used for the Blob client. Else authentication is done via account name and key using the
+   * {@link StorageSharedKeyCredential}.
+   * The config used to determine which authentication is systems.%s.azureblob.useTokenCredentialAuthentication = true
+   * for using TokenCredential.
+   * @return BlobServiceAsyncClient
+   */
+  public BlobServiceAsyncClient getBlobServiceAsyncClient() {
+    BlobServiceClientBuilder blobServiceClientBuilder = getBlobServiceClientBuilder();
+
+    if (azureBlobConfig.getUseTokenCredentialAuthentication(systemName)) {
+      // Use your Azure Blob Storage account's name and client details to create a token credential object to access your account.
+      TokenCredential tokenCredential = getTokenCredential();
+      return blobServiceClientBuilder.credential(tokenCredential).buildAsyncClient();
+    }
+
+    // Use your Azure Blob Storage account's name and key to create a credential object to access your account.
+    StorageSharedKeyCredential storageSharedKeyCredential = getStorageSharedKeyCredential();
+    return blobServiceClientBuilder.credential(storageSharedKeyCredential).buildAsyncClient();
+  }
+
+  /**
+   * Method to get the builder {@link BlobServiceClientBuilder} for creating BlobServiceAsyncClient.
+   * This builder is not provided the credential for authentication here.
+   * this builder is given an endpoint for the Azure Storage account and a http client to be passed on to the
+   * BlobServiceAsyncClient for blob creation.
+   * @return BlobServiceClientBuilder
+   */
+  private BlobServiceClientBuilder getBlobServiceClientBuilder() {
+    // From the Azure portal, get your Storage account blob service AsyncClient endpoint.
+    String endpoint = String.format(Locale.ROOT, azureUrlFormat, azureBlobConfig.getAzureAccountName(systemName));
+
+    HttpLogOptions httpLogOptions = new HttpLogOptions();
+    httpLogOptions.setLogLevel(HttpLogDetailLevel.BASIC);
+
+    BlobServiceClientBuilder blobServiceClientBuilder = new BlobServiceClientBuilder()
+        .httpLogOptions(httpLogOptions)
+        .endpoint(endpoint)
+        .httpClient(getHttpClient());
+
+    return blobServiceClientBuilder;
+  }
+
+  /**
+   * Method to create the {@link com.azure.identity.ClientSecretCredential} to be used for authenticating with
+   * Azure Storage. If the config systems.%s.azureblob.authProxy.use is set to true, then authentication happens
+   * via the proxy specified by systems.%s.azureblob.authProxy.hostname and systems.%s.azureblob.authProxy.port configs.
+   * @return ClientSecretCredential which extends from {@link TokenCredential}
+   */
+  private TokenCredential getTokenCredential() {
+    ClientSecretCredentialBuilder clientSecretCredentialBuilder = new ClientSecretCredentialBuilder()
+        .clientId(azureBlobConfig.getAzureClientId(systemName))
+        .clientSecret(azureBlobConfig.getAzureClientSecret(systemName))
+        .tenantId(azureBlobConfig.getAzureTenantId(systemName));
+
+    if (azureBlobConfig.getUseAuthProxy(systemName)) {
+      return clientSecretCredentialBuilder
+          .proxyOptions(new ProxyOptions(ProxyOptions.Type.HTTP,
+              new InetSocketAddress(azureBlobConfig.getAuthProxyHostName(systemName),
+                  azureBlobConfig.getAuthProxyPort(systemName))))
+          .build();
+    }
+    return clientSecretCredentialBuilder
+        .build();
+  }
+
+  /**
+   * Method to create {@link StorageSharedKeyCredential} to used for authenticating with Azure Storage.
+   * @return StorageSharedKeyCredential
+   */
+  private StorageSharedKeyCredential getStorageSharedKeyCredential() {
+    return new StorageSharedKeyCredential(azureBlobConfig.getAzureAccountName(systemName),
+        azureBlobConfig.getAzureAccountKey(systemName));
+  }
+
+  /**
+   * Method to create {@link HttpClient} to be used by the {@link BlobServiceAsyncClient} while creating blobs
+   * in the Azure Storage. If the config systems.%s.azureblob.proxy.use is set to true then the http client
+   * uses the proxy provided by systems.%s.azureblob.proxy.hostname and systems.%s.azureblob.proxy.port configs.
+   * @return HttpClient
+   */
+  private HttpClient getHttpClient() {
+    HttpClient httpClient;
+    if (azureBlobConfig.getUseBlobProxy(systemName)) {
+      httpClient = new NettyAsyncHttpClientBuilder()
+          .proxy(new ProxyOptions(ProxyOptions.Type.HTTP,
+              new InetSocketAddress(azureBlobConfig.getAzureBlobProxyHostname(systemName),
+                  azureBlobConfig.getAzureBlobProxyPort(systemName))))
+          .build();
+    } else {
+      httpClient = HttpClient.createDefault();
+    }
+    return httpClient;
+  }
+}
diff --git a/samza-azure/src/main/java/org/apache/samza/system/azureblob/AzureBlobConfig.java b/samza-azure/src/main/java/org/apache/samza/system/azureblob/AzureBlobConfig.java
index 58f206e..3026a89 100644
--- a/samza-azure/src/main/java/org/apache/samza/system/azureblob/AzureBlobConfig.java
+++ b/samza-azure/src/main/java/org/apache/samza/system/azureblob/AzureBlobConfig.java
@@ -33,11 +33,27 @@
   public static final String AZURE_BLOB_LOG_SLOW_REQUESTS_MS = "samza.azureblob.log.slowRequestMs";
   private static final long AZURE_BLOB_LOG_SLOW_REQUESTS_MS_DEFAULT = Duration.ofSeconds(30).toMillis();
 
+  // Azure authentication - via a/c name+key or ClientSecretCredential
   // system Level Properties.
   // fully qualified class name of the AzureBlobWriter impl for the producer system
   public static final String SYSTEM_WRITER_FACTORY_CLASS_NAME = SYSTEM_AZUREBLOB_PREFIX + "writer.factory.class";
   public static final String SYSTEM_WRITER_FACTORY_CLASS_NAME_DEFAULT = "org.apache.samza.system.azureblob.avro.AzureBlobAvroWriterFactory";
 
+  public static final String SYSTEM_USE_TOKEN_CREDENTIAL_AUTHENTICATION = Config.SENSITIVE_PREFIX + SYSTEM_AZUREBLOB_PREFIX + "useTokenCredentialAuthentication";
+  private static final boolean SYSTEM_USE_TOKEN_CREDENTIAL_AUTHENTICATION_DEFAULT = false;
+
+  // ClientSecretCredential needs client id, client secret, tenant id, vault name, service principal
+  public static final String SYSTEM_AZURE_CLIENT_ID = Config.SENSITIVE_PREFIX + SYSTEM_AZUREBLOB_PREFIX + "client.id";
+  public static final String SYSTEM_AZURE_CLIENT_SECRET = Config.SENSITIVE_PREFIX + SYSTEM_AZUREBLOB_PREFIX + "client.secret";
+  public static final String SYSTEM_AZURE_TENANT_ID = Config.SENSITIVE_PREFIX + SYSTEM_AZUREBLOB_PREFIX + "tenant.id";
+  // Whether to use proxy while authenticating with Azure
+  public static final String SYSTEM_AZURE_USE_AUTH_PROXY  = SYSTEM_AZUREBLOB_PREFIX + "authProxy.use";
+  public static final boolean SYSTEM_AZURE_USE_AUTH_PROXY_DEFAULT = false;
+
+  // name of the host to be used as auth proxy
+  public static final String SYSTEM_AZURE_AUTH_PROXY_HOSTNAME = SYSTEM_AZUREBLOB_PREFIX + "authProxy.hostname";
+  // port in the auth proxy host to be used
+  public static final String SYSTEM_AZURE_AUTH_PROXY_PORT = SYSTEM_AZUREBLOB_PREFIX + "authProxy.port";
   // Azure Storage Account name under which the Azure container representing this system is.
   // System name = Azure container name
   // (https://docs.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#container-names)
@@ -101,7 +117,6 @@
   public static final String SYSTEM_BLOB_METADATA_PROPERTIES_GENERATOR_FACTORY = SYSTEM_AZUREBLOB_PREFIX + "metadataPropertiesGeneratorFactory";
   private static final String SYSTEM_BLOB_METADATA_PROPERTIES_GENERATOR_FACTORY_DEFAULT =
       "org.apache.samza.system.azureblob.utils.NullBlobMetadataGeneratorFactory";
-
   // Additional configs for the metadata generator should be prefixed with this string which is passed to the generator.
   // for example, to pass a "key":"value" pair to the metadata generator, add config like
   // systems.<system-name>.azureblob.metadataGeneratorConfig.<key> with value <value>
@@ -127,11 +142,11 @@
     return accountName;
   }
 
-  public boolean getUseProxy(String systemName) {
+  public boolean getUseBlobProxy(String systemName) {
     return getBoolean(String.format(SYSTEM_AZURE_USE_PROXY, systemName), SYSTEM_AZURE_USE_PROXY_DEFAULT);
   }
 
-  public String getAzureProxyHostname(String systemName) {
+  public String getAzureBlobProxyHostname(String systemName) {
     String hostname = get(String.format(SYSTEM_AZURE_PROXY_HOSTNAME, systemName));
     if (hostname == null) {
       throw new ConfigException("Azure proxy host name is required.");
@@ -139,7 +154,7 @@
     return hostname;
   }
 
-  public int getAzureProxyPort(String systemName) {
+  public int getAzureBlobProxyPort(String systemName) {
     return getInt(String.format(SYSTEM_AZURE_PROXY_PORT, systemName));
   }
 
@@ -207,4 +222,50 @@
   public Config getSystemBlobMetadataGeneratorConfigs(String systemName) {
     return subset(String.format(SYSTEM_BLOB_METADATA_GENERATOR_CONFIG_PREFIX, systemName));
   }
+
+  public boolean getUseTokenCredentialAuthentication(String systemName) {
+    return getBoolean(String.format(SYSTEM_USE_TOKEN_CREDENTIAL_AUTHENTICATION, systemName),
+        SYSTEM_USE_TOKEN_CREDENTIAL_AUTHENTICATION_DEFAULT);
+  }
+
+  public boolean getUseAuthProxy(String systemName) {
+    return getBoolean(String.format(SYSTEM_AZURE_USE_AUTH_PROXY, systemName), SYSTEM_AZURE_USE_AUTH_PROXY_DEFAULT);
+  }
+
+  public String getAuthProxyHostName(String systemName) {
+    String hostname = get(String.format(SYSTEM_AZURE_AUTH_PROXY_HOSTNAME, systemName));
+    if (hostname == null) {
+      throw new ConfigException("Azure proxy host name is required.");
+    }
+    return hostname;
+  }
+
+  public int getAuthProxyPort(String systemName) {
+    return getInt(String.format(SYSTEM_AZURE_AUTH_PROXY_PORT, systemName));
+  }
+
+  public String getAzureClientId(String systemName) {
+    String clientId = get(String.format(SYSTEM_AZURE_CLIENT_ID, systemName));
+    if (clientId == null) {
+      throw new ConfigException("Azure Client id is required.");
+    }
+    return clientId;
+  }
+
+  public String getAzureClientSecret(String systemName) {
+    String clientSecret = get(String.format(SYSTEM_AZURE_CLIENT_SECRET, systemName));
+    if (clientSecret == null) {
+      throw new ConfigException("Azure Client secret is required.");
+    }
+    return clientSecret;
+  }
+
+
+  public String getAzureTenantId(String systemName) {
+    String tenantId = get(String.format(SYSTEM_AZURE_TENANT_ID, systemName));
+    if (tenantId == null) {
+      throw new ConfigException("Azure tenant id is required.");
+    }
+    return tenantId;
+  }
 }
diff --git a/samza-azure/src/main/java/org/apache/samza/system/azureblob/avro/AzureBlobOutputStream.java b/samza-azure/src/main/java/org/apache/samza/system/azureblob/avro/AzureBlobOutputStream.java
index e615808..222b99c 100644
--- a/samza-azure/src/main/java/org/apache/samza/system/azureblob/avro/AzureBlobOutputStream.java
+++ b/samza-azure/src/main/java/org/apache/samza/system/azureblob/avro/AzureBlobOutputStream.java
@@ -273,7 +273,7 @@
 
   // SAMZA-2476 stubbing BlockBlobAsyncClient.stageBlock was causing flaky tests.
   @VisibleForTesting
-  void stageBlock(String blockIdEncoded, ByteBuffer outputStream, int blockSize) {
+  void stageBlock(String blockIdEncoded, ByteBuffer outputStream, int blockSize) throws InterruptedException {
     blobAsyncClient.stageBlock(blockIdEncoded, Flux.just(outputStream), blockSize).block();
   }
 
@@ -335,6 +335,11 @@
             // StageBlock generates exception on Failure.
             stageBlock(blockIdEncoded, outputStream, blockSize);
             break;
+          } catch (InterruptedException e) {
+            String msg = String.format("Upload block for blob: %s failed for blockid: %s due to InterruptedException.",
+                blobAsyncClient.getBlobUrl().toString(), blockId);
+            LOG.error(msg, e);
+            throw new AzureException("InterruptedException encountered during block upload. Will not retry.", e);
           } catch (Exception e) {
             attemptCount += 1;
             String msg = "Upload block for blob: " + blobAsyncClient.getBlobUrl().toString()
diff --git a/samza-azure/src/main/java/org/apache/samza/system/azureblob/producer/AzureBlobSystemProducer.java b/samza-azure/src/main/java/org/apache/samza/system/azureblob/producer/AzureBlobSystemProducer.java
index d89f38f..781dce4 100644
--- a/samza-azure/src/main/java/org/apache/samza/system/azureblob/producer/AzureBlobSystemProducer.java
+++ b/samza-azure/src/main/java/org/apache/samza/system/azureblob/producer/AzureBlobSystemProducer.java
@@ -19,27 +19,16 @@
 
 package org.apache.samza.system.azureblob.producer;
 
-import com.azure.core.http.HttpClient;
 import com.azure.core.http.HttpResponse;
-import com.azure.core.http.ProxyOptions;
-import com.azure.core.http.netty.NettyAsyncHttpClientBuilder;
-import com.azure.core.http.policy.HttpLogDetailLevel;
-import com.azure.core.http.policy.HttpLogOptions;
 import com.azure.core.util.Configuration;
 import com.azure.storage.blob.BlobContainerAsyncClient;
 import com.azure.storage.blob.BlobServiceAsyncClient;
-import com.azure.storage.blob.BlobServiceClientBuilder;
 import com.azure.storage.blob.models.BlobErrorCode;
 import com.azure.storage.blob.models.BlobStorageException;
 import com.azure.storage.blob.models.SkuName;
-import com.azure.storage.common.StorageSharedKeyCredential;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import org.apache.samza.system.azureblob.AzureBlobConfig;
-import org.apache.samza.system.azureblob.compression.CompressionFactory;
 import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
@@ -55,6 +44,9 @@
 import org.apache.samza.system.OutgoingMessageEnvelope;
 import org.apache.samza.system.SystemProducer;
 import org.apache.samza.system.SystemProducerException;
+import org.apache.samza.system.azureblob.AzureBlobClientBuilder;
+import org.apache.samza.system.azureblob.AzureBlobConfig;
+import org.apache.samza.system.azureblob.compression.CompressionFactory;
 import org.apache.samza.system.azureblob.utils.BlobMetadataGeneratorFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -196,11 +188,7 @@
       LOG.warn("Attempting to start an already started producer.");
       return;
     }
-
-    String accountName = config.getAzureAccountName(systemName);
-    String accountKey = config.getAzureAccountKey(systemName);
-
-    setupAzureContainer(accountName, accountKey);
+    setupAzureContainer();
 
     LOG.info("Starting producer.");
     isStarted = true;
@@ -354,47 +342,11 @@
   }
 
   @VisibleForTesting
-  void setupAzureContainer(String accountName, String accountKey) {
+  void setupAzureContainer() {
     try {
-      // Use your Azure Blob Storage account's name and key to create a credential object to access your account.
-      StorageSharedKeyCredential credential = new StorageSharedKeyCredential(accountName, accountKey);
-
-      HttpClient httpClient;
-      if (config.getUseProxy(systemName)) {
-        LOG.info("HTTP Proxy setup for AzureBlob pipeline");
-        httpClient = new NettyAsyncHttpClientBuilder()
-            .proxy(new ProxyOptions(ProxyOptions.Type.HTTP,
-            new InetSocketAddress(config.getAzureProxyHostname(systemName), config.getAzureProxyPort(systemName)))).build();
-      } else {
-        httpClient = HttpClient.createDefault();
-      }
-
-      // From the Azure portal, get your Storage account blob service AsyncClient endpoint.
-      String endpoint = String.format(Locale.ROOT, AZURE_URL, accountName);
-
-      HttpLogOptions httpLogOptions = new HttpLogOptions();
-      httpLogOptions.setLogLevel(HttpLogDetailLevel.BASIC);
-      BlobServiceAsyncClient storageClient =
-          new BlobServiceClientBuilder()
-          .httpLogOptions(httpLogOptions)
-          .endpoint(endpoint)
-          .credential(credential)
-          .httpClient(httpClient)
-          .buildAsyncClient();
-
-
-      SkuName accountType = storageClient.getAccountInfo().block().getSkuName();
-      long flushThresholdSize = config.getMaxFlushThresholdSize(systemName);
-      boolean isPremiumAccount = SkuName.PREMIUM_LRS == accountType;
-      if (isPremiumAccount && flushThresholdSize > PREMIUM_MAX_BLOCK_SIZE) { // 100 MB
-        throw new SystemProducerException("Azure storage account with name: " + accountName
-            + " is a premium account and can only handle upto " +  PREMIUM_MAX_BLOCK_SIZE + " threshold size. Given flush threshold size is "
-            + flushThresholdSize);
-      } else if (!isPremiumAccount && flushThresholdSize > STANDARD_MAX_BLOCK_SIZE) { // STANDARD account
-        throw new SystemProducerException("Azure storage account with name: " + accountName
-            + " is a standard account and can only handle upto " + STANDARD_MAX_BLOCK_SIZE + " threshold size. Given flush threshold size is "
-            + flushThresholdSize);
-      }
+      BlobServiceAsyncClient storageClient  = new AzureBlobClientBuilder(systemName, AZURE_URL, config)
+          .getBlobServiceAsyncClient();
+      validateFlushThresholdSizeSupported(storageClient);
 
       containerAsyncClient = storageClient.getBlobContainerAsyncClient(systemName);
 
@@ -406,6 +358,22 @@
     }
   }
 
+  void validateFlushThresholdSizeSupported(BlobServiceAsyncClient storageClient) {
+    SkuName accountType = storageClient.getAccountInfo().block().getSkuName();
+    String accountName = storageClient.getAccountName();
+    long flushThresholdSize = config.getMaxFlushThresholdSize(systemName);
+    boolean isPremiumAccount = SkuName.PREMIUM_LRS == accountType;
+    if (isPremiumAccount && flushThresholdSize > PREMIUM_MAX_BLOCK_SIZE) { // 100 MB
+      throw new SystemProducerException("Azure storage account with name: " + accountName
+          + " is a premium account and can only handle upto " +  PREMIUM_MAX_BLOCK_SIZE + " threshold size. Given flush threshold size is "
+          + flushThresholdSize);
+    } else if (!isPremiumAccount && flushThresholdSize > STANDARD_MAX_BLOCK_SIZE) { // STANDARD account
+      throw new SystemProducerException(
+          "Azure storage account with name: " + accountName + " is a standard account and can only handle upto " + STANDARD_MAX_BLOCK_SIZE + " threshold size. Given flush threshold size is "
+              + flushThresholdSize);
+    }
+  }
+
   /**
    * // find the writer in the writerMap else create one
    * @param source for which to find/create the writer
diff --git a/samza-azure/src/test/java/org/apache/samza/system/azureblob/avro/TestAzureBlobOutputStream.java b/samza-azure/src/test/java/org/apache/samza/system/azureblob/avro/TestAzureBlobOutputStream.java
index b713ec7..4412edf 100644
--- a/samza-azure/src/test/java/org/apache/samza/system/azureblob/avro/TestAzureBlobOutputStream.java
+++ b/samza-azure/src/test/java/org/apache/samza/system/azureblob/avro/TestAzureBlobOutputStream.java
@@ -127,7 +127,7 @@
   }
 
   @Test
-  public void testWrite() {
+  public void testWrite() throws  InterruptedException {
     byte[] b = new byte[THRESHOLD - 10];
     azureBlobOutputStream.write(b, 0, THRESHOLD - 10);
     verify(azureBlobOutputStream, never()).stageBlock(anyString(), any(ByteBuffer.class), anyInt());
@@ -136,7 +136,7 @@
   }
 
   @Test
-  public void testWriteLargerThanThreshold() {
+  public void testWriteLargerThanThreshold() throws  InterruptedException {
     byte[] largeRecord = RANDOM_STRING.substring(0, 2 * THRESHOLD).getBytes();
     byte[] largeRecordFirstHalf = RANDOM_STRING.substring(0, THRESHOLD).getBytes();
     byte[] largeRecordSecondHalf = RANDOM_STRING.substring(THRESHOLD, 2 * THRESHOLD).getBytes();
@@ -165,7 +165,7 @@
   }
 
   @Test
-  public void testWriteLargeRecordWithSmallRecordInBuffer() {
+  public void testWriteLargeRecordWithSmallRecordInBuffer() throws InterruptedException {
     byte[] halfBlock = new byte[THRESHOLD / 2];
     byte[] fullBlock = new byte[THRESHOLD];
     byte[] largeRecord = new byte[2 * THRESHOLD];
@@ -229,6 +229,36 @@
     azureBlobOutputStream.close();
   }
 
+  @Test(expected = AzureException.class)
+  public void testWriteFailedInterruptedException() throws InterruptedException {
+
+    doThrow(new InterruptedException("Lets interrupt the thread"))
+        .when(azureBlobOutputStream).stageBlock(anyString(), any(ByteBuffer.class), anyInt());
+    byte[] b = new byte[100];
+    doReturn(COMPRESSED_BYTES).when(mockCompression).compress(b);
+
+    try {
+      azureBlobOutputStream.write(b, 0, THRESHOLD); // threshold crossed so stageBlock is scheduled.
+      // azureBlobOutputStream.close waits on the CompletableFuture which does the actual stageBlock in uploadBlockAsync
+      azureBlobOutputStream.close();
+    } catch (AzureException exception) {
+      // get root cause of the exception - to confirm its an InterruptedException
+      Throwable dupException = exception;
+      while (dupException.getCause() != null && dupException.getCause() != dupException) {
+        dupException = dupException.getCause();
+      }
+
+      Assert.assertTrue(dupException.getClass().getName().equals(InterruptedException.class.getCanonicalName()));
+      Assert.assertEquals("Lets interrupt the thread", dupException.getMessage());
+
+      // verify stageBlock was called exactly once - aka no retries happen when interrupted exception is thrown
+      verify(azureBlobOutputStream).stageBlock(anyString(), any(ByteBuffer.class), anyInt());
+
+      // rethrow the exception so that the test will fail if no exception was thrown in the try block
+      throw exception;
+    }
+  }
+
   @Test
   public void testClose() {
     azureBlobOutputStream.write(BYTES, 0, THRESHOLD);
@@ -278,7 +308,7 @@
   }
 
   @Test(expected = AzureException.class)
-  public void testCloseFailed() {
+  public void testCloseFailed() throws InterruptedException {
 
     azureBlobOutputStream = spy(new AzureBlobOutputStream(mockBlobAsyncClient, threadPool, mockMetrics,
         blobMetadataGeneratorFactory, blobMetadataGeneratorConfig, FAKE_STREAM,
@@ -320,7 +350,7 @@
   }
 
   @Test (expected = AzureException.class)
-  public void testFlushFailed() throws IOException {
+  public void testFlushFailed() throws IOException, InterruptedException {
     azureBlobOutputStream = spy(new AzureBlobOutputStream(mockBlobAsyncClient, threadPool, mockMetrics,
         blobMetadataGeneratorFactory, blobMetadataGeneratorConfig, FAKE_STREAM,
         60000, THRESHOLD, mockByteArrayOutputStream, mockCompression));
diff --git a/samza-azure/src/test/java/org/apache/samza/system/azureblob/producer/TestAzureBlobSystemProducer.java b/samza-azure/src/test/java/org/apache/samza/system/azureblob/producer/TestAzureBlobSystemProducer.java
index acf4cfb..a364cf6 100644
--- a/samza-azure/src/test/java/org/apache/samza/system/azureblob/producer/TestAzureBlobSystemProducer.java
+++ b/samza-azure/src/test/java/org/apache/samza/system/azureblob/producer/TestAzureBlobSystemProducer.java
@@ -98,14 +98,14 @@
     // use mock writer impl
     setupWriterForProducer(systemProducer, mockAzureWriter, STREAM);
     // bypass Azure connection setup
-    doNothing().when(systemProducer).setupAzureContainer(anyString(), anyString());
+    doNothing().when(systemProducer).setupAzureContainer();
   }
 
   @Test
   public void testStart() {
 
     systemProducer.start();
-    verify(systemProducer).setupAzureContainer(ACCOUNT_NAME, ACCOUNT_KEY);
+    verify(systemProducer).setupAzureContainer();
   }
 
   public void testMultipleStart() {
@@ -264,7 +264,7 @@
         mockMetricsRegistry));
     PowerMockito.whenNew(AzureBlobAvroWriter.class).withAnyArguments().thenThrow(new SystemProducerException("Failed"));
     // bypass Azure connection setup
-    doNothing().when(systemProducer).setupAzureContainer(anyString(), anyString());
+    doNothing().when(systemProducer).setupAzureContainer();
 
     systemProducer.register(SOURCE);
     systemProducer.start();
@@ -315,7 +315,7 @@
 
     AzureBlobSystemProducer systemProducer = spy(new AzureBlobSystemProducer(SYSTEM_NAME, azureBlobConfig, mockMetricsRegistry));
     // bypass Azure connection setup
-    doNothing().when(systemProducer).setupAzureContainer(anyString(), anyString());
+    doNothing().when(systemProducer).setupAzureContainer();
 
     doReturn(mockAzureWriter1).when(systemProducer).getOrCreateWriter(source1, ome1);
     doReturn(mockAzureWriter2).when(systemProducer).getOrCreateWriter(source2, ome2);
@@ -359,7 +359,7 @@
 
     AzureBlobSystemProducer systemProducer = spy(new AzureBlobSystemProducer(SYSTEM_NAME, azureBlobConfig, mockMetricsRegistry));
     // bypass Azure connection setup
-    doNothing().when(systemProducer).setupAzureContainer(anyString(), anyString());
+    doNothing().when(systemProducer).setupAzureContainer();
 
     doReturn(mockAzureWriter1).when(systemProducer).getOrCreateWriter(source1, ome1);
     doReturn(mockAzureWriter2).when(systemProducer).getOrCreateWriter(source2, ome2);
@@ -411,7 +411,7 @@
 
     AzureBlobSystemProducer systemProducer = spy(new AzureBlobSystemProducer(SYSTEM_NAME, azureBlobConfig, mockMetricsRegistry));
     // bypass Azure connection setup
-    doNothing().when(systemProducer).setupAzureContainer(anyString(), anyString());
+    doNothing().when(systemProducer).setupAzureContainer();
 
     systemProducer.register(source1);
     systemProducer.start();
@@ -450,7 +450,7 @@
 
     AzureBlobSystemProducer systemProducer = spy(new AzureBlobSystemProducer(SYSTEM_NAME, azureBlobConfig, mockMetricsRegistry));
     // bypass Azure connection setup
-    doNothing().when(systemProducer).setupAzureContainer(anyString(), anyString());
+    doNothing().when(systemProducer).setupAzureContainer();
 
     setupWriterForProducer(systemProducer, mockAzureWriter1, stream1);
 
@@ -492,7 +492,7 @@
 
     AzureBlobSystemProducer systemProducer = spy(new AzureBlobSystemProducer(SYSTEM_NAME, azureBlobConfig, mockMetricsRegistry));
     // bypass Azure connection setup
-    doNothing().when(systemProducer).setupAzureContainer(anyString(), anyString());
+    doNothing().when(systemProducer).setupAzureContainer();
 
     setupWriterForProducer(systemProducer, mockAzureWriter1, STREAM);
 
diff --git a/samza-core/src/main/java/org/apache/samza/classloader/DependencyIsolationUtils.java b/samza-core/src/main/java/org/apache/samza/classloader/DependencyIsolationUtils.java
deleted file mode 100644
index a0b5d1e..0000000
--- a/samza-core/src/main/java/org/apache/samza/classloader/DependencyIsolationUtils.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.classloader;
-
-public class DependencyIsolationUtils {
-  /**
-   * Directory inside the home directory of the cluster-based job coordinator in which the framework API artifacts are
-   * placed, for usage in dependency isolation for the cluster-based job coordinator.
-   * TODO make this configurable or taken from an environment variable
-   */
-  public static final String FRAMEWORK_API_DIRECTORY = "__samzaFrameworkApi";
-
-  /**
-   * Directory inside the home directory of the cluster-based job coordinator in which the framework infrastructure
-   * artifacts are placed, for usage in dependency isolation for the cluster-based job coordinator.
-   * TODO make this configurable or taken from an environment variable
-   */
-  public static final String FRAMEWORK_INFRASTRUCTURE_DIRECTORY = "__samzaFrameworkInfrastructure";
-
-  /**
-   * Directory inside the home directory of the cluster-based job coordinator in which the application artifacts are
-   * placed, for usage in dependency isolation for the cluster-based job coordinator.
-   * TODO make this configurable or taken from an environment variable
-   */
-  public static final String APPLICATION_DIRECTORY = "__package";
-
-  /**
-   * Name of the file which contains the class names (or globs) which should be loaded from the framework API
-   * classloader.
-   */
-  public static final String FRAMEWORK_API_CLASS_LIST_FILE_NAME = "samza-framework-api-classes.txt";
-
-  public static final String RUNTIME_FRAMEWORK_RESOURCES_PATHING_JAR_NAME = "runtime-framework-resources-pathing.jar";
-}
diff --git a/samza-core/src/main/java/org/apache/samza/classloader/IsolatingClassLoaderFactory.java b/samza-core/src/main/java/org/apache/samza/classloader/IsolatingClassLoaderFactory.java
deleted file mode 100644
index 344a034..0000000
--- a/samza-core/src/main/java/org/apache/samza/classloader/IsolatingClassLoaderFactory.java
+++ /dev/null
@@ -1,352 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.classloader;
-
-import com.linkedin.cytodynamics.matcher.BootstrapClassPredicate;
-import com.linkedin.cytodynamics.matcher.GlobMatcher;
-import com.linkedin.cytodynamics.nucleus.DelegateRelationship;
-import com.linkedin.cytodynamics.nucleus.DelegateRelationshipBuilder;
-import com.linkedin.cytodynamics.nucleus.IsolationLevel;
-import com.linkedin.cytodynamics.nucleus.LoaderBuilder;
-import com.linkedin.cytodynamics.nucleus.OriginRestriction;
-import java.io.File;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.net.URL;
-import java.net.URLClassLoader;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import com.google.common.annotations.VisibleForTesting;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.samza.SamzaException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * Use this to build a classloader for running Samza which isolates the Samza framework code/dependencies from the
- * application code/dependencies.
- */
-public class IsolatingClassLoaderFactory {
-  private static final Logger LOG = LoggerFactory.getLogger(IsolatingClassLoaderFactory.class);
-
-  private static final String LIB_DIRECTORY = "lib";
-
-  /**
-   * Build a classloader which will isolate Samza framework code from application code. Samza framework classes and
-   * application-specific classes will be loaded using a different classloaders. This will enable dependencies of each
-   * category of classes to also be loaded separately, so that runtime dependency conflicts do not happen.
-   * Each call to this method will build a different instance of a classloader.
-   *
-   * Samza framework API classes need to be specified in a file called
-   * {@link DependencyIsolationUtils#FRAMEWORK_API_CLASS_LIST_FILE_NAME} which is in the lib directory which is in the
-   * API package. The file needs to be generated when building the framework API package. This class will not generate
-   * the file.
-   *
-   * Implementation notes:
-   *
-   * The cytodynamics isolating classloader is used for this. It provides more control than the built-in
-   * {@link URLClassLoader}. Cytodynamics provides the ability to compose multiple classloaders together and have more
-   * granular delegation strategies between the classloaders.
-   *
-   * In order to share objects between classes loaded by different classloaders, the classes for the shared objects must
-   * be loaded by a common classloader. Those common classes will be loaded through a common API classloader. The
-   * cytodynamics classloader can be set up to only use the common API classloader for an explicit set of classes. The
-   * {@link DependencyIsolationUtils#FRAMEWORK_API_CLASS_LIST_FILE_NAME} file should include the framework API classes.
-   * Also, bootstrap classes (e.g. java.lang.String) need to be loaded by a common classloader, since objects of those
-   * types need to be shared across different framework and application. There are also some static bootstrap classes
-   * which should be shared (e.g. java.lang.System). Bootstrap classes will be loaded through a common classloader by
-   * default.
-   *
-   * These are the classloaders which are used to make up the final classloader.
-   * <ul>
-   *   <li>bootstrap classloader: Built-in Java classes (e.g. java.lang.String)</li>
-   *   <li>API classloader: Common Samza framework API classes</li>
-   *   <li>infrastructure classloader: Core Samza framework classes and plugins that are included in the framework</li>
-   *   <li>
-   *     application classloader: Application code and plugins that are needed in the app but are not included in the
-   *     framework
-   *   </li>
-   * </ul>
-   *
-   * This is the delegation structure for the classloaders:
-   * <pre>
-   *   (bootstrap               (API                  (application
-   *   classloader) &lt;---- classloader) &lt;------- classloader)
-   *                             ^                      ^
-   *                             |                     /
-   *                             |                    /
-   *                             |                   /
-   *                             |                  /
-   *                         (infrastructure classloader)
-   * </pre>
-   * The cytodynamics classloader allows control over when the delegation should happen.
-   * <ol>
-   *   <li>API classloader delegates to the bootstrap classloader if the bootstrap classloader has the class.</li>
-   *   <li>
-   *     Infrastructure classloader only delegates to the API classloader for the common classes specified by
-   *     {@link DependencyIsolationUtils#FRAMEWORK_API_CLASS_LIST_FILE_NAME}.
-   *   </li>
-   *   <li>
-   *     Infrastructure classloader delegates to the application classloader when a class can't be found in the
-   *     infrastructure classloader.
-   *   </li>
-   *   <li>
-   *     Application classloader only delegates to the API classloader for the common classes specified by
-   *     {@link DependencyIsolationUtils#FRAMEWORK_API_CLASS_LIST_FILE_NAME}.
-   *   </li>
-   * </ol>
-   */
-  public ClassLoader buildClassLoader() {
-    // start at the user.dir to find the resources for the classpaths
-    File baseJobDirectory = new File(System.getProperty("user.dir"));
-    File apiLibDirectory = libDirectory(new File(baseJobDirectory, DependencyIsolationUtils.FRAMEWORK_API_DIRECTORY));
-    LOG.info("Using API lib directory: {}", apiLibDirectory);
-    File infrastructureLibDirectory =
-        libDirectory(new File(baseJobDirectory, DependencyIsolationUtils.FRAMEWORK_INFRASTRUCTURE_DIRECTORY));
-    LOG.info("Using infrastructure lib directory: {}", infrastructureLibDirectory);
-    File applicationLibDirectory =
-        libDirectory(new File(baseJobDirectory, DependencyIsolationUtils.APPLICATION_DIRECTORY));
-    LOG.info("Using application lib directory: {}", applicationLibDirectory);
-
-    ClassLoader apiClassLoader = buildApiClassLoader(apiLibDirectory);
-    ClassLoader applicationClassLoader =
-        buildApplicationClassLoader(applicationLibDirectory, apiLibDirectory, apiClassLoader);
-
-    // the classloader to return is the one with the infrastructure classpath
-    return buildInfrastructureClassLoader(infrastructureLibDirectory, baseJobDirectory, apiLibDirectory, apiClassLoader,
-        applicationClassLoader);
-  }
-
-  /**
-   * Build the {@link ClassLoader} which can load framework API classes.
-   *
-   * This sets up the link between the bootstrap classloader and the API classloader (see {@link #buildClassLoader()}.
-   */
-  private static ClassLoader buildApiClassLoader(File apiLibDirectory) {
-    /*
-     * This can just use the built-in classloading, which checks the parent classloader first and then checks its own
-     * classpath. A null parent means bootstrap classloader, which contains core Java classes (e.g. java.lang.String).
-     * This doesn't need to be isolated from the parent, because we only want to load all bootstrap classes from the
-     * bootstrap classloader.
-     */
-    return new URLClassLoader(getClasspathAsURLs(apiLibDirectory), null);
-  }
-
-  /**
-   * Build the {@link ClassLoader} which can load application classes.
-   *
-   * This sets up the link between the application classloader and the API classloader (see {@link #buildClassLoader()}.
-   */
-  private static ClassLoader buildApplicationClassLoader(File applicationLibDirectory, File apiLibDirectory,
-      ClassLoader apiClassLoader) {
-    return LoaderBuilder.anIsolatingLoader()
-        // look in application lib directory for JARs
-        .withClasspath(getClasspathAsURIs(applicationLibDirectory))
-        // getClasspathAsURIs should only return JARs within applicationLibDirectory anyways, but doing it to be safe
-        .withOriginRestriction(OriginRestriction.denyByDefault().allowingDirectory(applicationLibDirectory, false))
-        // delegate to the api classloader for API classes
-        .withParentRelationship(buildApiParentRelationship(apiLibDirectory, apiClassLoader))
-        .build();
-  }
-
-  /**
-   * Build the {@link ClassLoader} which can load Samza framework core classes. If a file with the name
-   * {@link DependencyIsolationUtils#RUNTIME_FRAMEWORK_RESOURCES_PATHING_JAR_NAME} is found in {@code baseJobDirectory},
-   * then it will be included in the classpath.
-   * This may also fall back to loading application classes.
-   *
-   * This sets up two links: One link between the infrastructure classloader and the API and another link between the
-   * infrastructure classloader and the application classloader (see {@link #buildClassLoader()}.
-   */
-  private static ClassLoader buildInfrastructureClassLoader(File infrastructureLibDirectory,
-      File baseJobDirectory,
-      File apiLibDirectory,
-      ClassLoader apiClassLoader,
-      ClassLoader applicationClassLoader) {
-    // start with JARs in infrastructure lib directory
-    List<URI> classpathURIs = new ArrayList<>(getClasspathAsURIs(infrastructureLibDirectory));
-    OriginRestriction originRestriction = OriginRestriction.denyByDefault()
-        // getClasspathAsURIs should only return JARs within infrastructureLibDirectory anyways, but doing it to be safe
-        .allowingDirectory(infrastructureLibDirectory, false);
-    File runtimeFrameworkResourcesPathingJar =
-        new File(baseJobDirectory, DependencyIsolationUtils.RUNTIME_FRAMEWORK_RESOURCES_PATHING_JAR_NAME);
-    if (canAccess(runtimeFrameworkResourcesPathingJar)) {
-      // if there is a runtime framework resources pathing JAR, then include that in the classpath as well
-      classpathURIs.add(runtimeFrameworkResourcesPathingJar.toURI());
-      originRestriction.allowingGlobPattern(fileURL(runtimeFrameworkResourcesPathingJar).toExternalForm());
-      LOG.info("Added {} to infrastructure classpath", runtimeFrameworkResourcesPathingJar.getPath());
-    } else {
-      LOG.info("Unable to access {}, so not adding to infrastructure classpath",
-          runtimeFrameworkResourcesPathingJar.getPath());
-    }
-    return LoaderBuilder.anIsolatingLoader()
-        .withClasspath(Collections.unmodifiableList(classpathURIs))
-        .withOriginRestriction(originRestriction)
-        .withParentRelationship(buildApiParentRelationship(apiLibDirectory, apiClassLoader))
-        /*
-         * Fall back to the application classloader for certain classes. For example, the application might implement
-         * some pluggable classes (e.g. SystemFactory). Another example is message schemas that are supplied by the
-         * application.
-         */
-        .addFallbackDelegate(DelegateRelationshipBuilder.builder()
-            .withDelegateClassLoader(applicationClassLoader)
-            /*
-             * NONE means that a class will be loaded from here if it is not found in the classpath of the loader that uses
-             * this relationship.
-             */
-            .withIsolationLevel(IsolationLevel.NONE)
-            .build())
-        .build();
-  }
-
-  /**
-   * Build a {@link DelegateRelationship} which defines how to delegate to the API classloader.
-   *
-   * Delegation will only happen for classes specified in
-   * {@link DependencyIsolationUtils#FRAMEWORK_API_CLASS_LIST_FILE_NAME} and the Java bootstrap classes.
-   */
-  private static DelegateRelationship buildApiParentRelationship(File apiLibDirectory, ClassLoader apiClassLoader) {
-    DelegateRelationshipBuilder apiParentRelationshipBuilder = DelegateRelationshipBuilder.builder()
-        // needs to load API classes from the API classloader
-        .withDelegateClassLoader(apiClassLoader)
-        /*
-         * FULL means to only load classes explicitly specified as "API" from the API classloader. We will use
-         * delegate-preferred class predicates to specify which classes are "API" (see below).
-         */
-        .withIsolationLevel(IsolationLevel.FULL);
-
-    // bootstrap classes need to be loaded from a common classloader
-    apiParentRelationshipBuilder.addDelegatePreferredClassPredicate(new BootstrapClassPredicate());
-    // the classes which are Samza framework API classes are added here
-    getFrameworkApiClassGlobs(apiLibDirectory).forEach(
-      apiClassName -> apiParentRelationshipBuilder.addDelegatePreferredClassPredicate(new GlobMatcher(apiClassName)));
-    return apiParentRelationshipBuilder.build();
-  }
-
-  /**
-   * Gets the globs for matching against classes to load from the framework API classloader. This will read the
-   * {@link DependencyIsolationUtils#FRAMEWORK_API_CLASS_LIST_FILE_NAME} file in {@code directoryWithClassList} to get
-   * the globs.
-   *
-   * @param directoryWithClassList Directory in which
-   * {@link DependencyIsolationUtils#FRAMEWORK_API_CLASS_LIST_FILE_NAME} lives
-   * @return {@link List} of globs for matching against classes to load from the framework API classloader
-   */
-  @VisibleForTesting
-  static List<String> getFrameworkApiClassGlobs(File directoryWithClassList) {
-    File parentPreferredFile =
-        new File(directoryWithClassList, DependencyIsolationUtils.FRAMEWORK_API_CLASS_LIST_FILE_NAME);
-    validateCanAccess(parentPreferredFile);
-    try {
-      return Files.readAllLines(Paths.get(parentPreferredFile.toURI()), StandardCharsets.UTF_8)
-          .stream()
-          .filter(StringUtils::isNotBlank)
-          .collect(Collectors.toList());
-    } catch (IOException e) {
-      throw new SamzaException("Error while reading samza-api class list", e);
-    }
-  }
-
-  /**
-   * Get the {@link URL}s of all JARs/WARs in the directory {@code jarsLocation}. This only looks one level down; it is
-   * not recursive.
-   */
-  @VisibleForTesting
-  static URL[] getClasspathAsURLs(File jarsLocation) {
-    validateCanAccess(jarsLocation);
-    File[] filesInJarsLocation = jarsLocation.listFiles();
-    if (filesInJarsLocation == null) {
-      throw new SamzaException(
-          String.format("Could not find any files inside %s, probably because it is not a directory",
-              jarsLocation.getPath()));
-    }
-    URL[] urls = Stream.of(filesInJarsLocation)
-        .filter(file -> file.getName().endsWith(".jar") || file.getName().endsWith(".war"))
-        .map(IsolatingClassLoaderFactory::fileURL)
-        .toArray(URL[]::new);
-    LOG.info("Found {} items to load into classpath from {}", urls.length, jarsLocation);
-    Stream.of(urls).forEach(url -> LOG.debug("Found {} from {}", url, jarsLocation));
-    return urls;
-  }
-
-  /**
-   * Get the {@link URI}s of all JARs/WARs in the directory {@code jarsLocation}. This only looks one level down; it is
-   * not recursive.
-   */
-  @VisibleForTesting
-  static List<URI> getClasspathAsURIs(File jarsLocation) {
-    return Stream.of(getClasspathAsURLs(jarsLocation))
-        .map(IsolatingClassLoaderFactory::urlToURI)
-        .collect(Collectors.toList());
-  }
-
-  private static boolean canAccess(File file) {
-    return file.exists() && file.canRead();
-  }
-
-  /**
-   * Makes sure that a file exists and can be read.
-   */
-  private static void validateCanAccess(File file) {
-    if (!canAccess(file)) {
-      throw new SamzaException("Unable to access file: " + file);
-    }
-  }
-
-  /**
-   * Get the {@link URL} for a {@link File}.
-   * Converts checked exceptions into {@link SamzaException}s.
-   */
-  private static URL fileURL(File file) {
-    URI uri = file.toURI();
-    try {
-      return uri.toURL();
-    } catch (MalformedURLException e) {
-      throw new SamzaException("Unable to get URL for file: " + file, e);
-    }
-  }
-
-  /**
-   * Get the {@link URI} for a {@link URL}.
-   * Converts checked exceptions into {@link SamzaException}s.
-   */
-  private static URI urlToURI(URL url) {
-    try {
-      return url.toURI();
-    } catch (URISyntaxException e) {
-      throw new SamzaException("Unable to get URI for URL: " + url, e);
-    }
-  }
-
-  /**
-   * Get the {@link File} representing the {@link #LIB_DIRECTORY} inside the given {@code file}.
-   */
-  private static File libDirectory(File file) {
-    return new File(file, LIB_DIRECTORY);
-  }
-}
diff --git a/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinatorRunner.java b/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinatorRunner.java
index a152032..523742c 100644
--- a/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinatorRunner.java
+++ b/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinatorRunner.java
@@ -26,7 +26,6 @@
 import java.util.List;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.samza.SamzaException;
-import org.apache.samza.classloader.IsolatingClassLoaderFactory;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.MapConfig;
@@ -35,7 +34,6 @@
 import org.apache.samza.metrics.MetricsRegistryMap;
 import org.apache.samza.serializers.model.SamzaObjectMapper;
 import org.apache.samza.util.CoordinatorStreamUtil;
-import org.apache.samza.util.SplitDeploymentUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -52,19 +50,12 @@
       LOG.error("Uncaught exception in ClusterBasedJobCoordinator::main. Exiting job coordinator", exception);
       System.exit(1);
     });
-    if (!SplitDeploymentUtil.isSplitDeploymentEnabled()) {
-      // no isolation enabled, so can just execute runClusterBasedJobCoordinator directly
-      runClusterBasedJobCoordinator(args);
-    } else {
-      SplitDeploymentUtil.runWithClassLoader(new IsolatingClassLoaderFactory().buildClassLoader(),
-          ClusterBasedJobCoordinatorRunner.class, "runClusterBasedJobCoordinator", args);
-    }
+    runClusterBasedJobCoordinator(args);
     System.exit(0);
   }
 
   /**
-   * This is the actual execution for the {@link ClusterBasedJobCoordinator}. This is separated out from
-   * {@link #main(String[])} so that it can be executed directly or from a separate classloader.
+   * This is the actual execution for the {@link ClusterBasedJobCoordinator}.
    */
   @VisibleForTesting
   static void runClusterBasedJobCoordinator(String[] args) {
diff --git a/samza-core/src/main/java/org/apache/samza/clustermanager/SingleFaultDomainManager.java b/samza-core/src/main/java/org/apache/samza/clustermanager/SingleFaultDomainManager.java
new file mode 100644
index 0000000..34a5aad
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/clustermanager/SingleFaultDomainManager.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.clustermanager;
+
+import java.util.Collections;
+import java.util.Set;
+
+
+/**
+ * Simple placeholder implementation of {@link FaultDomainManager} which contains a single fault domain for all hosts.
+ * This can be used when another concrete {@link FaultDomainManager} is undesirable or unavailable, but features which
+ * depend on a {@link FaultDomainManager} (such as standby containers) may have unexpected behavior.
+ */
+public class SingleFaultDomainManager implements FaultDomainManager {
+  private static final FaultDomain SINGLE_FAULT_DOMAIN = new FaultDomain(FaultDomainType.RACK, "0");
+
+  @Override
+  public Set<FaultDomain> getAllFaultDomains() {
+    return Collections.singleton(SINGLE_FAULT_DOMAIN);
+  }
+
+  @Override
+  public Set<FaultDomain> getFaultDomainsForHost(String host) {
+    return Collections.singleton(SINGLE_FAULT_DOMAIN);
+  }
+
+  @Override
+  public boolean hasSameFaultDomains(String host1, String host2) {
+    return true;
+  }
+}
diff --git a/samza-core/src/main/java/org/apache/samza/clustermanager/SingleFaultDomainManagerFactory.java b/samza-core/src/main/java/org/apache/samza/clustermanager/SingleFaultDomainManagerFactory.java
new file mode 100644
index 0000000..db96b15
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/clustermanager/SingleFaultDomainManagerFactory.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.clustermanager;
+
+import org.apache.samza.config.Config;
+import org.apache.samza.metrics.MetricsRegistry;
+
+
+/**
+ * Produces a simple placeholder {@link FaultDomainManager}. See {@link SingleFaultDomainManager}.
+ */
+public class SingleFaultDomainManagerFactory implements FaultDomainManagerFactory {
+  @Override
+  public FaultDomainManager getFaultDomainManager(Config config, MetricsRegistry metricsRegistry) {
+    return new SingleFaultDomainManager();
+  }
+}
diff --git a/samza-core/src/main/java/org/apache/samza/config/ClusterManagerConfig.java b/samza-core/src/main/java/org/apache/samza/config/ClusterManagerConfig.java
index 3f27991..acb5a0a 100644
--- a/samza-core/src/main/java/org/apache/samza/config/ClusterManagerConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/ClusterManagerConfig.java
@@ -137,6 +137,20 @@
   private static final String AM_JMX_ENABLED = "yarn.am.jmx.enabled";
   private static final String CLUSTER_MANAGER_JMX_ENABLED = "cluster-manager.jobcoordinator.jmx.enabled";
 
+  /**
+   * Use this to configure a static port for the job coordinator url for a Samza job. This url is used to provide
+   * information such as job model and locality.
+   * If the value is set to 0, then the port will be dynamically allocated from the available free ports on the node.
+   * The default value of this config is 0.
+   *
+   * Be careful when using this configuration. If the configured port is already in use on the node, then the job
+   * coordinator will fail to start.
+   *
+   * This configuration is experimental, and it might be removed in a future release.
+   */
+  private static final String JOB_COORDINATOR_URL_PORT = "cluster-manager.jobcoordinator.url.port";
+  private static final int DEFAULT_JOB_COORDINATOR_URL_PORT = 0;
+
   public ClusterManagerConfig(Config config) {
       super(config);
   }
@@ -280,4 +294,8 @@
       return true;
     }
   }
+
+  public int getCoordinatorUrlPort() {
+    return getInt(JOB_COORDINATOR_URL_PORT, DEFAULT_JOB_COORDINATOR_URL_PORT);
+  }
 }
diff --git a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
index 12257e0..4822067 100644
--- a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java
@@ -155,8 +155,6 @@
   public static final String COORDINATOR_STREAM_FACTORY = "job.coordinatorstream.config.factory";
   public static final String DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY = "org.apache.samza.util.DefaultCoordinatorStreamConfigFactory";
 
-  public static final String JOB_SPLIT_DEPLOYMENT_ENABLED = "job.split.deployment.enabled";
-
   private static final String JOB_STARTPOINT_ENABLED = "job.startpoint.enabled";
 
   // Enable ClusterBasedJobCoordinator aka ApplicationMaster High Availability (AM-HA).
@@ -414,10 +412,6 @@
     return getStandbyTaskReplicationFactor() > 1;
   }
 
-  public boolean isSplitDeploymentEnabled() {
-    return getBoolean(JOB_SPLIT_DEPLOYMENT_ENABLED, false);
-  }
-
   /**
    * The metadata file is written in a {@code exec-env-container-id}.metadata file in the log-dir of the container.
    * Here the {@code exec-env-container-id} refers to the ID assigned by the cluster manager (e.g., YARN) to the container,
diff --git a/samza-core/src/main/java/org/apache/samza/config/ShellCommandConfig.java b/samza-core/src/main/java/org/apache/samza/config/ShellCommandConfig.java
index 73093a8..73bcf8e 100644
--- a/samza-core/src/main/java/org/apache/samza/config/ShellCommandConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/ShellCommandConfig.java
@@ -57,29 +57,6 @@
    */
   public static final String ENV_EXECUTION_ENV_CONTAINER_ID = "EXECUTION_ENV_CONTAINER_ID";
 
-  /**
-   * Set to "true" if split deployment feature is enabled. Otherwise, will be considered false.
-   *
-   * The launch process for the cluster-based job coordinator and job container depends on the value of this, since it
-   * needs to be known if the cluster-based job coordinator and job container should be launched in a split deployment
-   * mode.
-   * This needs to be an environment variable, because the value needs to be known before the full configs can be read
-   * from the metadata store (full configs are only read after launch is complete).
-   */
-  public static final String ENV_SPLIT_DEPLOYMENT_ENABLED = "ENV_SPLIT_DEPLOYMENT_ENABLED";
-
-  /**
-   * When running the cluster-based job coordinator and job container in a split deployment mode, it uses JARs and
-   * resources from a lib directory which is provided by the framework. In some cases, it is necessary to use some
-   * resources specified by the application as well. This environment variable can be set to a directory which is
-   * different from the framework lib directory in order to tell Samza where application resources live.
-   * This is an environment variable because it is needed in order to launch the cluster-based job coordinator and job
-   * container Java processes, which means access to full configs is not available yet.
-   * For example, this is used to set a system property for the location of an application-specified log4j configuration
-   * file when launching the cluster-based job coordinator and job container Java processes.
-   */
-  public static final String ENV_APPLICATION_LIB_DIR = "APPLICATION_LIB_DIR";
-
   /*
    * The base directory for storing logged data stores used in Samza. This has to be set on all machine running Samza
    * containers. For example, when using YARN, it has to be set in all NMs and passed to the containers.
diff --git a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java
index 4367244..b170bac 100644
--- a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java
@@ -262,17 +262,6 @@
   }
 
   /**
-   * Helper method to check if a system has a changelog attached to it.
-   */
-  public boolean isChangelogSystem(String systemName) {
-    return getStoreNames().stream()
-        .map(this::getChangelogStream)
-        .filter(Optional::isPresent)
-        .map(systemStreamName -> StreamUtil.getSystemStreamFromNames(systemStreamName.get()).getSystem())
-        .anyMatch(system -> system.equals(systemName));
-  }
-
-  /**
    * Helper method to check if there is any stores configured w/ a changelog
    */
   public boolean hasDurableStores() {
diff --git a/samza-core/src/main/java/org/apache/samza/context/InternalTaskContext.java b/samza-core/src/main/java/org/apache/samza/context/InternalTaskContext.java
index 3d3a53d..804ef93 100644
--- a/samza-core/src/main/java/org/apache/samza/context/InternalTaskContext.java
+++ b/samza-core/src/main/java/org/apache/samza/context/InternalTaskContext.java
@@ -21,12 +21,24 @@
 
 import org.apache.samza.job.model.JobModel;
 import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemStreamPartition;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 
+
+/**
+ * This class is used for passing objects around for the implementation of the high-level API.
+ * 1) Container for objects that need to be passed between different components in the implementation of the high-level
+ * API.
+ * 2) The implementation of the high-level API is built on top of the low-level API. The low-level API only exposes
+ * {@link TaskContext}, but the implementation of the high-level API needs some other internal Samza components (e.g.
+ * {@link StreamMetadataCache}. We internally make these components available through {@link TaskContextImpl} so that we
+ * can do a cast to access the components. This class hides some of the messiness of casting. It's still not ideal to
+ * need to do any casting, even in this class.
+ */
 public class InternalTaskContext {
-
   private final Context context;
   private final Map<String, Object> objectRegistry = new HashMap<>();
 
@@ -46,6 +58,10 @@
     return context;
   }
 
+  /**
+   * TODO: The public {@link JobContext} exposes {@link JobModel} now, so can this internal method be replaced by the
+   * public API?
+   */
   public JobModel getJobModel() {
     return ((TaskContextImpl) this.context.getTaskContext()).getJobModel();
   }
@@ -53,4 +69,11 @@
   public StreamMetadataCache getStreamMetadataCache() {
     return ((TaskContextImpl) this.context.getTaskContext()).getStreamMetadataCache();
   }
+
+  /**
+   * See {@link TaskContextImpl#getSspsExcludingSideInputs()}.
+   */
+  public Set<SystemStreamPartition> getSspsExcludingSideInputs() {
+    return ((TaskContextImpl) this.context.getTaskContext()).getSspsExcludingSideInputs();
+  }
 }
diff --git a/samza-core/src/main/java/org/apache/samza/context/TaskContextImpl.java b/samza-core/src/main/java/org/apache/samza/context/TaskContextImpl.java
index edec17d..d87a5bc 100644
--- a/samza-core/src/main/java/org/apache/samza/context/TaskContextImpl.java
+++ b/samza-core/src/main/java/org/apache/samza/context/TaskContextImpl.java
@@ -29,9 +29,15 @@
 import org.apache.samza.table.ReadWriteTable;
 import org.apache.samza.table.TableManager;
 
+import java.util.Set;
 import java.util.function.Function;
 
 
+/**
+ * This class provides the implementation for the public {@link TaskContext} interface.
+ * It also allows us to pass certain internal Samza components around so that the implementation of the high-level API
+ * can use them (see InternalTaskContext for some more details).
+ */
 public class TaskContextImpl implements TaskContext {
   private final TaskModel taskModel;
   private final MetricsRegistry taskMetricsRegistry;
@@ -39,8 +45,13 @@
   private final TableManager tableManager;
   private final CallbackScheduler callbackScheduler;
   private final OffsetManager offsetManager;
+
+  // The instance variables below are not used for implementing any public API methods. They are here so that we can
+  // pass some internal components over to the implementation of the high-level API. See InternalTaskContext.
+
   private final JobModel jobModel;
   private final StreamMetadataCache streamMetadataCache;
+  private final Set<SystemStreamPartition> sspsExcludingSideInputs;
 
   public TaskContextImpl(TaskModel taskModel,
       MetricsRegistry taskMetricsRegistry,
@@ -49,7 +60,8 @@
       CallbackScheduler callbackScheduler,
       OffsetManager offsetManager,
       JobModel jobModel,
-      StreamMetadataCache streamMetadataCache) {
+      StreamMetadataCache streamMetadataCache,
+      Set<SystemStreamPartition> sspsExcludingSideInputs) {
     this.taskModel = taskModel;
     this.taskMetricsRegistry = taskMetricsRegistry;
     this.keyValueStoreProvider = keyValueStoreProvider;
@@ -58,6 +70,7 @@
     this.offsetManager = offsetManager;
     this.jobModel = jobModel;
     this.streamMetadataCache = streamMetadataCache;
+    this.sspsExcludingSideInputs = sspsExcludingSideInputs;
   }
 
   @Override
@@ -101,4 +114,14 @@
   public StreamMetadataCache getStreamMetadataCache() {
     return this.streamMetadataCache;
   }
-}
+
+  /**
+   * Returns the {@link SystemStreamPartition}s excluding the side-input SSPs. For the high-level API, watermarks and
+   * end-of-stream messages are propagated based on their input SSPs. However, the Samza framework does not give side
+   * input messages to the high-level operator tasks. Therefore, the operators need to know the input SSPs excluding the
+   * side input SSPs. See SAMZA-2303 for more details.
+   */
+  public Set<SystemStreamPartition> getSspsExcludingSideInputs() {
+    return this.sspsExcludingSideInputs;
+  }
+}
\ No newline at end of file
diff --git a/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorMetadataManager.java b/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorMetadataManager.java
index 860c596..7e40382 100644
--- a/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorMetadataManager.java
+++ b/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorMetadataManager.java
@@ -36,7 +36,6 @@
 import org.apache.samza.job.JobCoordinatorMetadata;
 import org.apache.samza.job.model.JobModel;
 import org.apache.samza.metadatastore.MetadataStore;
-import org.apache.samza.metrics.Counter;
 import org.apache.samza.metrics.Gauge;
 import org.apache.samza.metrics.MetricsRegistry;
 import org.apache.samza.serializers.Serde;
@@ -284,44 +283,44 @@
     private static final String METADATA_WRITE_FAILED_COUNT = "metadata-write-failed-count";
     private static final String NEW_DEPLOYMENT = "new-deployment";
 
-    private final Counter applicationAttemptCount;
-    private final Counter metadataGenerationFailedCount;
-    private final Counter metadataReadFailedCount;
-    private final Counter metadataWriteFailedCount;
+    private final Gauge<Integer> applicationAttemptCount;
+    private final Gauge<Integer> metadataGenerationFailedCount;
+    private final Gauge<Integer> metadataReadFailedCount;
+    private final Gauge<Integer> metadataWriteFailedCount;
     private final Gauge<Integer> jobModelChangedAcrossApplicationAttempt;
     private final Gauge<Integer> configChangedAcrossApplicationAttempt;
     private final Gauge<Integer> newDeployment;
 
     public JobCoordinatorMetadataManagerMetrics(MetricsRegistry registry) {
-      applicationAttemptCount = registry.newCounter(GROUP, APPLICATION_ATTEMPT_COUNT);
+      applicationAttemptCount = registry.newGauge(GROUP, APPLICATION_ATTEMPT_COUNT, 0);
       configChangedAcrossApplicationAttempt =
           registry.newGauge(GROUP, CONFIG_CHANGED, 0);
       jobModelChangedAcrossApplicationAttempt =
           registry.newGauge(GROUP, JOB_MODEL_CHANGED, 0);
-      metadataGenerationFailedCount = registry.newCounter(GROUP,
-          METADATA_GENERATION_FAILED_COUNT);
-      metadataReadFailedCount = registry.newCounter(GROUP, METADATA_READ_FAILED_COUNT);
-      metadataWriteFailedCount = registry.newCounter(GROUP, METADATA_WRITE_FAILED_COUNT);
+      metadataGenerationFailedCount = registry.newGauge(GROUP,
+          METADATA_GENERATION_FAILED_COUNT, 0);
+      metadataReadFailedCount = registry.newGauge(GROUP, METADATA_READ_FAILED_COUNT, 0);
+      metadataWriteFailedCount = registry.newGauge(GROUP, METADATA_WRITE_FAILED_COUNT, 0);
       newDeployment = registry.newGauge(GROUP, NEW_DEPLOYMENT, 0);
     }
 
     @VisibleForTesting
-    Counter getApplicationAttemptCount() {
+    Gauge<Integer> getApplicationAttemptCount() {
       return applicationAttemptCount;
     }
 
     @VisibleForTesting
-    Counter getMetadataGenerationFailedCount() {
+    Gauge<Integer> getMetadataGenerationFailedCount() {
       return metadataGenerationFailedCount;
     }
 
     @VisibleForTesting
-    Counter getMetadataReadFailedCount() {
+    Gauge<Integer> getMetadataReadFailedCount() {
       return metadataReadFailedCount;
     }
 
     @VisibleForTesting
-    Counter getMetadataWriteFailedCount() {
+    Gauge<Integer> getMetadataWriteFailedCount() {
       return metadataWriteFailedCount;
     }
 
@@ -341,19 +340,19 @@
     }
 
     void incrementApplicationAttemptCount() {
-      applicationAttemptCount.inc();
+      applicationAttemptCount.set(applicationAttemptCount.getValue() + 1);
     }
 
     void incrementMetadataGenerationFailedCount() {
-      metadataGenerationFailedCount.inc();
+      metadataGenerationFailedCount.set(metadataGenerationFailedCount.getValue() + 1);
     }
 
     void incrementMetadataReadFailedCount() {
-      metadataReadFailedCount.inc();
+      metadataReadFailedCount.set(metadataReadFailedCount.getValue() + 1);
     }
 
     void incrementMetadataWriteFailedCount() {
-      metadataWriteFailedCount.inc();
+      metadataWriteFailedCount.set(metadataWriteFailedCount.getValue() + 1);
     }
 
     void setConfigChangedAcrossApplicationAttempt(int value) {
diff --git a/samza-core/src/main/java/org/apache/samza/job/ShellCommandBuilder.java b/samza-core/src/main/java/org/apache/samza/job/ShellCommandBuilder.java
new file mode 100644
index 0000000..3725344
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/job/ShellCommandBuilder.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.job;
+
+import java.io.File;
+import java.util.Map;
+import com.google.common.collect.ImmutableMap;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.samza.config.ShellCommandConfig;
+
+
+public class ShellCommandBuilder extends CommandBuilder {
+  @Override
+  public String buildCommand() {
+    ShellCommandConfig shellCommandConfig = new ShellCommandConfig(config);
+    if (StringUtils.isEmpty(this.commandPath)) {
+      return shellCommandConfig.getCommand();
+    } else {
+      return this.commandPath + File.separator + shellCommandConfig.getCommand();
+    }
+  }
+
+  @Override
+  public Map<String, String> buildEnvironment() {
+    ShellCommandConfig shellCommandConfig = new ShellCommandConfig(config);
+    ImmutableMap.Builder<String, String> envBuilder = new ImmutableMap.Builder<>();
+    envBuilder.put(ShellCommandConfig.ENV_CONTAINER_ID, this.id);
+    envBuilder.put(ShellCommandConfig.ENV_COORDINATOR_URL, this.url.toString());
+    envBuilder.put(ShellCommandConfig.ENV_JAVA_OPTS, shellCommandConfig.getTaskOpts().orElse(""));
+    envBuilder.put(ShellCommandConfig.ENV_ADDITIONAL_CLASSPATH_DIR,
+        shellCommandConfig.getAdditionalClasspathDir().orElse(""));
+    shellCommandConfig.getJavaHome().ifPresent(javaHome -> envBuilder.put(ShellCommandConfig.ENV_JAVA_HOME, javaHome));
+    return envBuilder.build();
+  }
+}
diff --git a/samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporter.java b/samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporter.java
new file mode 100644
index 0000000..5961d7a
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporter.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.metrics.reporter;
+
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.samza.metrics.Counter;
+import org.apache.samza.metrics.Gauge;
+import org.apache.samza.metrics.Metric;
+import org.apache.samza.metrics.MetricsReporter;
+import org.apache.samza.metrics.MetricsVisitor;
+import org.apache.samza.metrics.ReadableMetricsRegistry;
+import org.apache.samza.metrics.Timer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Implementation of {@link MetricsReporter} which logs metrics which match a regex.
+ * The regex is checked against "[source name]-[group name]-[metric name]".
+ */
+public class LoggingMetricsReporter implements MetricsReporter {
+  private static final Logger LOG = LoggerFactory.getLogger(LoggingMetricsReporter.class);
+  /**
+   * First part is source, second part is group name, third part is metric name
+   */
+  private static final String FULL_METRIC_FORMAT = "%s-%s-%s";
+
+  private final ScheduledExecutorService scheduledExecutorService;
+  private final Pattern metricsToLog;
+  private final long loggingIntervalSeconds;
+  private final Queue<Runnable> loggingTasks = new ConcurrentLinkedQueue<>();
+
+  /**
+   * @param scheduledExecutorService executes the logging tasks
+   * @param metricsToLog Only log the metrics which match this regex. The strings for matching against this metric are
+   *                     constructed by concatenating source name, group name, and metric name, delimited by dashes.
+   * @param loggingIntervalSeconds interval at which to log metrics
+   */
+  public LoggingMetricsReporter(ScheduledExecutorService scheduledExecutorService, Pattern metricsToLog,
+      long loggingIntervalSeconds) {
+    this.scheduledExecutorService = scheduledExecutorService;
+    this.metricsToLog = metricsToLog;
+    this.loggingIntervalSeconds = loggingIntervalSeconds;
+  }
+
+  @Override
+  public void start() {
+    this.scheduledExecutorService.scheduleAtFixedRate(() -> this.loggingTasks.forEach(Runnable::run),
+        this.loggingIntervalSeconds, this.loggingIntervalSeconds, TimeUnit.SECONDS);
+  }
+
+  @Override
+  public void register(String source, ReadableMetricsRegistry registry) {
+    this.loggingTasks.add(buildLoggingTask(source, registry));
+  }
+
+  @Override
+  public void stop() {
+    this.scheduledExecutorService.shutdown();
+    try {
+      this.scheduledExecutorService.awaitTermination(10, TimeUnit.SECONDS);
+    } catch (InterruptedException e) {
+      LOG.warn("Interrupted while shutting down executor", e);
+    }
+    if (!this.scheduledExecutorService.isTerminated()) {
+      LOG.warn("Unable to shutdown executor");
+    }
+  }
+
+  /**
+   * VisibleForTesting so that the logging call can be verified in unit tests.
+   */
+  @VisibleForTesting
+  void doLog(String logString) {
+    LOG.info(logString);
+  }
+
+  private Runnable buildLoggingTask(String source, ReadableMetricsRegistry registry) {
+    return () -> {
+      for (String group : registry.getGroups()) {
+        for (Map.Entry<String, Metric> metricGroupEntry : registry.getGroup(group).entrySet()) {
+          metricGroupEntry.getValue().visit(new MetricsVisitor() {
+            @Override
+            public void counter(Counter counter) {
+              logMetric(source, group, counter.getName(), counter.getCount());
+            }
+
+            @Override
+            public <T> void gauge(Gauge<T> gauge) {
+              logMetric(source, group, gauge.getName(), gauge.getValue());
+            }
+
+            @Override
+            public void timer(Timer timer) {
+              logMetric(source, group, timer.getName(), timer.getSnapshot().getAverage());
+            }
+          });
+        }
+      }
+    };
+  }
+
+  private <T> void logMetric(String source, String group, String metricName, T value) {
+    String fullMetricName = String.format(FULL_METRIC_FORMAT, source, group, metricName);
+    if (this.metricsToLog.matcher(fullMetricName).matches()) {
+      doLog(String.format("Metric: %s, Value: %s", fullMetricName, value));
+    }
+  }
+}
diff --git a/samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporterConfig.java b/samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporterConfig.java
new file mode 100644
index 0000000..e7a256f
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporterConfig.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.metrics.reporter;
+
+import java.util.Optional;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.MapConfig;
+
+
+public class LoggingMetricsReporterConfig extends MapConfig {
+  private static final String METRICS_TO_LOG_REGEX_CONFIG = "metrics.reporter.%s.log.regex";
+  private static final String LOGGING_INTERVAL_SECONDS_CONFIG = "metrics.reporter.%s.logging.interval.seconds";
+  private static final long LOGGING_INTERVAL_SECONDS_DEFAULT = 60;
+
+  public LoggingMetricsReporterConfig(Config config) {
+    super(config);
+  }
+
+  public String getMetricsToLogRegex(String reporterName) {
+    String metricsToLogConfigKey = String.format(METRICS_TO_LOG_REGEX_CONFIG, reporterName);
+    return Optional.ofNullable(get(metricsToLogConfigKey))
+        .orElseThrow(() -> new ConfigException("Missing value for " + metricsToLogConfigKey));
+  }
+
+  public long getLoggingIntervalSeconds(String reporterName) {
+    return getLong(String.format(LOGGING_INTERVAL_SECONDS_CONFIG, reporterName), LOGGING_INTERVAL_SECONDS_DEFAULT);
+  }
+}
diff --git a/samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporterFactory.java b/samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporterFactory.java
new file mode 100644
index 0000000..88d3892
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporterFactory.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.metrics.reporter;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.regex.Pattern;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.samza.config.Config;
+import org.apache.samza.metrics.MetricsReporter;
+import org.apache.samza.metrics.MetricsReporterFactory;
+
+
+/**
+ * Creates a {@link MetricsReporter} which logs metrics and their values.
+ * This can be used to access metric values when no other external metrics system is available.
+ */
+public class LoggingMetricsReporterFactory implements MetricsReporterFactory {
+  @Override
+  public MetricsReporter getMetricsReporter(String name, String processorId, Config config) {
+    ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
+        new ThreadFactoryBuilder().setNameFormat("Samza LoggingMetricsReporter Thread-%d").setDaemon(true).build());
+    LoggingMetricsReporterConfig loggingMetricsReporterConfig = new LoggingMetricsReporterConfig(config);
+    Pattern metricsToLog = Pattern.compile(loggingMetricsReporterConfig.getMetricsToLogRegex(name));
+    return new LoggingMetricsReporter(scheduledExecutorService, metricsToLog,
+        loggingMetricsReporterConfig.getLoggingIntervalSeconds(name));
+  }
+}
diff --git a/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java b/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java
index 705f0cb..19cec80 100644
--- a/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java
+++ b/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java
@@ -111,17 +111,13 @@
       LOG.info("{} has {} producer tasks.", stream, count);
     });
 
-    // set states for end-of-stream
+    // set states for end-of-stream; don't include side inputs (see SAMZA-2303)
     internalTaskContext.registerObject(EndOfStreamStates.class.getName(),
-        new EndOfStreamStates(
-                internalTaskContext.getContext().getTaskContext().getTaskModel().getSystemStreamPartitions(),
-                producerTaskCounts));
-    // set states for watermark
+        new EndOfStreamStates(internalTaskContext.getSspsExcludingSideInputs(), producerTaskCounts));
+    // set states for watermark; don't include side inputs (see SAMZA-2303)
     internalTaskContext.registerObject(WatermarkStates.class.getName(),
-        new WatermarkStates(
-                internalTaskContext.getContext().getTaskContext().getTaskModel().getSystemStreamPartitions(),
-                producerTaskCounts,
-                context.getContainerContext().getContainerMetricsRegistry()));
+        new WatermarkStates(internalTaskContext.getSspsExcludingSideInputs(), producerTaskCounts,
+            context.getContainerContext().getContainerMetricsRegistry()));
 
     specGraph.getInputOperators().forEach((streamId, inputOpSpec) -> {
       SystemStream systemStream = streamConfig.streamIdToSystemStream(streamId);
diff --git a/samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java b/samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java
index bc4b227..87429ed 100644
--- a/samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java
+++ b/samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java
@@ -68,7 +68,7 @@
     List<IncomingMessageEnvelope> messages = bufferedMessages.get(ssp);
     String offset = String.valueOf(messages.size());
 
-    if (message instanceof EndOfStreamMessage) {
+    if (shouldUseEndOfStreamOffset(message)) {
       offset = IncomingMessageEnvelope.END_OF_STREAM_OFFSET;
     }
 
@@ -224,4 +224,22 @@
 
     return ImmutableList.copyOf(messageEnvelopesForSSP.subList(startingOffset, messageEnvelopesForSSP.size()));
   }
+
+  /**
+   * We don't always want to use {@link IncomingMessageEnvelope#END_OF_STREAM_OFFSET} for all
+   * {@link EndOfStreamMessage}s. Certain control message flows (e.g. end-of-stream) have an aggregation partition,
+   * which needs to listen for messages from all other partitions. These aggregation messages are marked by the task
+   * name being non-null. If we use {@link IncomingMessageEnvelope#END_OF_STREAM_OFFSET} for the aggregation messages,
+   * then the aggregation partition would stop listening once it got the message from one of the tasks, but that means
+   * it would miss the aggregation messages from all other tasks. See SAMZA-2300 for more details.
+   * One other note: If there is a serializer set for the stream, then by the time the message gets to this check, it
+   * will be a byte array, so this check will not return true, even if the deserialized message was an
+   * {@link EndOfStreamMessage}. So far this isn't a problem, because we only really need this to return true for
+   * input streams (not intermediate streams), and in-memory input stream data doesn't get serialized. For intermediate
+   * streams, we don't need END_OF_STREAM_OFFSET to be used since the high-level operators take care of end-of-stream
+   * messages based on MessageType.
+   */
+  private static boolean shouldUseEndOfStreamOffset(Object message) {
+    return (message instanceof EndOfStreamMessage) && ((EndOfStreamMessage) message).getTaskName() == null;
+  }
 }
diff --git a/samza-core/src/main/java/org/apache/samza/util/SplitDeploymentUtil.java b/samza-core/src/main/java/org/apache/samza/util/SplitDeploymentUtil.java
deleted file mode 100644
index 200cd3c..0000000
--- a/samza-core/src/main/java/org/apache/samza/util/SplitDeploymentUtil.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.util;
-
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import org.apache.samza.SamzaException;
-import org.apache.samza.config.ShellCommandConfig;
-
-
-public final class SplitDeploymentUtil {
-
-  /**
-   * The split deployment feature uses system env {@code ShellCommandConfig.ENV_SPLIT_DEPLOYMENT_ENABLED} to represent
-   * if the user chooses to enable it.
-   * This function helps to detect if the split deployment feature is enabled.
-   *
-   * @return true if split deployment is enabled; vice versa
-   */
-  public static boolean isSplitDeploymentEnabled() {
-    return Boolean.parseBoolean(System.getenv(ShellCommandConfig.ENV_SPLIT_DEPLOYMENT_ENABLED));
-  }
-
-  /**
-   * Execute the runner class using a separate isolated classloader.
-   * @param classLoader {@link ClassLoader} to use to load the runner class which will run
-   * @param originalRunnerClass {@link Class} for which will be executed with the new class loader.
-   * @param runMethodName run method name of runner class
-   * @param runMethodArgs arguments to pass to run method
-   */
-  public static void runWithClassLoader(ClassLoader classLoader, Class<?> originalRunnerClass, String runMethodName,
-      String[] runMethodArgs) {
-    // need to use the isolated classloader to load run method and then execute using that new class
-    Class<?> runnerClass;
-    try {
-      runnerClass = classLoader.loadClass(originalRunnerClass.getName());
-    } catch (ClassNotFoundException e) {
-      throw new SamzaException(String.format(
-          "Isolation was enabled, but unable to find %s in isolated classloader", originalRunnerClass.getName()), e);
-    }
-
-    // save the current context classloader so it can be reset after finishing the call to run method
-    ClassLoader previousContextClassLoader = Thread.currentThread().getContextClassLoader();
-    // this is needed because certain libraries (e.g. log4j) use the context classloader
-    Thread.currentThread().setContextClassLoader(classLoader);
-
-    try {
-      executeRunForRunnerClass(runnerClass, runMethodName, runMethodArgs);
-    } finally {
-      // reset the context class loader; it's good practice, and could be important when running a test suite
-      Thread.currentThread().setContextClassLoader(previousContextClassLoader);
-    }
-  }
-
-  private static void executeRunForRunnerClass(Class<?> runnerClass, String runMethodName, String[] runMethodArgs) {
-    Method runMethod;
-    try {
-      runMethod = runnerClass.getDeclaredMethod(runMethodName, String[].class);
-    } catch (NoSuchMethodException e) {
-      throw new SamzaException(String.format("Isolation was enabled, but unable to find %s method", runMethodName), e);
-    }
-    // only sets accessible flag for this method instance
-    runMethod.setAccessible(true);
-
-    try {
-      // wrapping args in object array so that args is passed as a single argument to the method
-      runMethod.invoke(null, new Object[]{runMethodArgs});
-    } catch (IllegalAccessException | InvocationTargetException e) {
-      throw new SamzaException(String.format("Exception while executing %s method", runMethodName), e);
-    }
-  }
-}
diff --git a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
index f91aae1..4e1c41a 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
@@ -88,7 +88,7 @@
       }
     })
   private val taskContext = new TaskContextImpl(taskModel, metrics.registry, kvStoreSupplier, tableManager,
-    new CallbackSchedulerImpl(epochTimeScheduler), offsetManager, jobModel, streamMetadataCache)
+    new CallbackSchedulerImpl(epochTimeScheduler), offsetManager, jobModel, streamMetadataCache, systemStreamPartitions)
   // need separate field for this instead of using it through Context, since Context throws an exception if it is null
   private val applicationTaskContextOption = applicationTaskContextFactoryOption
     .map(_.create(externalContextOption.orNull, jobContext, containerContext, taskContext,
diff --git a/samza-core/src/main/scala/org/apache/samza/coordinator/JobModelManager.scala b/samza-core/src/main/scala/org/apache/samza/coordinator/JobModelManager.scala
index 7c0e747..5dd662b 100644
--- a/samza-core/src/main/scala/org/apache/samza/coordinator/JobModelManager.scala
+++ b/samza-core/src/main/scala/org/apache/samza/coordinator/JobModelManager.scala
@@ -100,7 +100,8 @@
 
       updateTaskAssignments(jobModel, taskAssignmentManager, taskPartitionAssignmentManager, grouperMetadata)
 
-      val server = new HttpServer
+      val clusterManagerConfig = new ClusterManagerConfig(config)
+      val server = new HttpServer(port = clusterManagerConfig.getCoordinatorUrlPort)
       server.addServlet("/", new JobServlet(serializedJobModelRef))
       server.addServlet("/locality", new LocalityServlet(localityManager))
 
diff --git a/samza-core/src/main/scala/org/apache/samza/job/ShellCommandBuilder.scala b/samza-core/src/main/scala/org/apache/samza/job/ShellCommandBuilder.scala
deleted file mode 100644
index 9b95648..0000000
--- a/samza-core/src/main/scala/org/apache/samza/job/ShellCommandBuilder.scala
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.samza.job
-
-
-import java.io.File
-
-import org.apache.samza.config.ShellCommandConfig
-import org.apache.samza.util.ScalaJavaUtil.JavaOptionals
-
-import scala.collection.JavaConverters._
-
-class ShellCommandBuilder extends CommandBuilder {
-  def buildCommand() = {
-    val shellCommandConfig = new ShellCommandConfig(config)
-    if(commandPath == null || commandPath.isEmpty())
-      shellCommandConfig.getCommand
-    else
-      commandPath + File.separator +  shellCommandConfig.getCommand
-  }
-
-  def buildEnvironment(): java.util.Map[String, String] = {
-    val shellCommandConfig = new ShellCommandConfig(config)
-    val envMap = Map(
-      ShellCommandConfig.ENV_CONTAINER_ID -> id.toString,
-      ShellCommandConfig.ENV_COORDINATOR_URL -> url.toString,
-      ShellCommandConfig.ENV_JAVA_OPTS -> shellCommandConfig.getTaskOpts.orElse(""),
-      ShellCommandConfig.ENV_ADDITIONAL_CLASSPATH_DIR -> shellCommandConfig.getAdditionalClasspathDir.orElse(""))
-
-    val envMapWithJavaHome = JavaOptionals.toRichOptional(shellCommandConfig.getJavaHome).toOption match {
-      case Some(javaHome) => envMap + (ShellCommandConfig.ENV_JAVA_HOME -> javaHome)
-      case None => envMap
-    }
-
-    envMapWithJavaHome.asJava
-  }
-}
diff --git a/samza-core/src/test/java/org/apache/samza/classloader/TestIsolatingClassLoaderFactory.java b/samza-core/src/test/java/org/apache/samza/classloader/TestIsolatingClassLoaderFactory.java
deleted file mode 100644
index 7444fbf..0000000
--- a/samza-core/src/test/java/org/apache/samza/classloader/TestIsolatingClassLoaderFactory.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.classloader;
-
-import java.io.File;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.net.URL;
-import java.nio.file.Paths;
-import java.util.List;
-import java.util.Set;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
-import org.apache.samza.SamzaException;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-
-public class TestIsolatingClassLoaderFactory {
-  @Test
-  public void testGetApiClasses() throws URISyntaxException {
-    File apiClassListFile = Paths.get(getClass().getResource("/classloader").toURI()).toFile();
-    List<String> apiClassNames = IsolatingClassLoaderFactory.getFrameworkApiClassGlobs(apiClassListFile);
-    List<String> expected = ImmutableList.of(
-        "org.apache.samza.JavaClass",
-        "org.apache.samza.JavaClass$InnerJavaClass",
-        "org.apache.samza.ScalaClass$",
-        "org.apache.samza.ScalaClass$$anon$1",
-        "my.package.with.wildcard.*",
-        "my.package.with.question.mark?");
-    assertEquals(expected, apiClassNames);
-  }
-
-  @Test(expected = SamzaException.class)
-  public void testGetApiClassesFileDoesNotExist() throws URISyntaxException {
-    File nonExistentDirectory =
-        new File(Paths.get(getClass().getResource("/classloader").toURI()).toFile(), "doesNotExist");
-    IsolatingClassLoaderFactory.getFrameworkApiClassGlobs(nonExistentDirectory);
-  }
-
-  @Test
-  public void testGetClasspathAsURLs() throws URISyntaxException {
-    File classpathDirectory = Paths.get(getClass().getResource("/classloader/classpath").toURI()).toFile();
-    URL[] classpath = IsolatingClassLoaderFactory.getClasspathAsURLs(classpathDirectory);
-    assertEquals(2, classpath.length);
-    Set<URL> classpathSet = ImmutableSet.copyOf(classpath);
-    URL jarUrl = getClass().getResource("/classloader/classpath/placeholder-jar.jar");
-    assertTrue(classpathSet.contains(jarUrl));
-    URL warUrl = getClass().getResource("/classloader/classpath/placeholder-war.war");
-    assertTrue(classpathSet.contains(warUrl));
-  }
-
-  @Test(expected = SamzaException.class)
-  public void testGetClasspathAsURLsDirectoryDoesNotExist() throws URISyntaxException {
-    File nonExistentDirectory =
-        new File(Paths.get(getClass().getResource("/classloader").toURI()).toFile(), "doesNotExist");
-    IsolatingClassLoaderFactory.getClasspathAsURLs(nonExistentDirectory);
-  }
-
-  @Test
-  public void testGetClasspathAsURIs() throws URISyntaxException {
-    File classpathDirectory = Paths.get(getClass().getResource("/classloader/classpath").toURI()).toFile();
-    List<URI> classpath = IsolatingClassLoaderFactory.getClasspathAsURIs(classpathDirectory);
-    assertEquals(2, classpath.size());
-    Set<URI> classpathSet = ImmutableSet.copyOf(classpath);
-    URL jarUrl = getClass().getResource("/classloader/classpath/placeholder-jar.jar");
-    assertTrue(classpathSet.contains(jarUrl.toURI()));
-    URL warUrl = getClass().getResource("/classloader/classpath/placeholder-war.war");
-    assertTrue(classpathSet.contains(warUrl.toURI()));
-  }
-
-  @Test(expected = SamzaException.class)
-  public void testGetClasspathAsURIsDirectoryDoesNotExist() throws URISyntaxException {
-    File nonExistentDirectory =
-        new File(Paths.get(getClass().getResource("/classloader").toURI()).toFile(), "doesNotExist");
-    IsolatingClassLoaderFactory.getClasspathAsURIs(nonExistentDirectory);
-  }
-}
\ No newline at end of file
diff --git a/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java b/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java
index 00c2d5e..af4bd1d 100644
--- a/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java
+++ b/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java
@@ -546,17 +546,6 @@
   }
 
   @Test
-  public void testGetClusterBasedJobCoordinatorDependencyIsolationEnabled() {
-    Config config = new MapConfig(ImmutableMap.of(JobConfig.JOB_SPLIT_DEPLOYMENT_ENABLED, "true"));
-    assertTrue(new JobConfig(config).isSplitDeploymentEnabled());
-
-    config = new MapConfig(ImmutableMap.of(JobConfig.JOB_SPLIT_DEPLOYMENT_ENABLED, "false"));
-    assertFalse(new JobConfig(config).isSplitDeploymentEnabled());
-
-    assertFalse(new JobConfig(new MapConfig()).isSplitDeploymentEnabled());
-  }
-
-  @Test
   public void testGetMetadataFile() {
     String execEnvContainerId = "container-id";
     String containerMetadataDirectory = "/tmp/samza/log/dir";
diff --git a/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java b/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java
index 5dba698..0d06c3b 100644
--- a/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java
+++ b/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java
@@ -359,18 +359,6 @@
   }
 
   @Test
-  public void testIsChangelogSystem() {
-    StorageConfig storageConfig = new StorageConfig(new MapConfig(ImmutableMap.of(
-        // store0 has a changelog stream
-        String.format(StorageConfig.FACTORY, STORE_NAME0), "factory.class",
-        String.format(CHANGELOG_STREAM, STORE_NAME0), "system0.changelog-stream",
-        // store1 does not have a changelog stream
-        String.format(StorageConfig.FACTORY, STORE_NAME1), "factory.class")));
-    assertTrue(storageConfig.isChangelogSystem("system0"));
-    assertFalse(storageConfig.isChangelogSystem("other-system"));
-  }
-
-  @Test
   public void testHasDurableStores() {
     // no changelog, which means no durable stores
     StorageConfig storageConfig = new StorageConfig(
diff --git a/samza-core/src/test/java/org/apache/samza/context/TestTaskContextImpl.java b/samza-core/src/test/java/org/apache/samza/context/TestTaskContextImpl.java
index 0e8f78e..094583e 100644
--- a/samza-core/src/test/java/org/apache/samza/context/TestTaskContextImpl.java
+++ b/samza-core/src/test/java/org/apache/samza/context/TestTaskContextImpl.java
@@ -62,7 +62,7 @@
     MockitoAnnotations.initMocks(this);
     taskContext =
         new TaskContextImpl(taskModel, taskMetricsRegistry, keyValueStoreProvider, tableManager, callbackScheduler,
-            offsetManager, null, null);
+            offsetManager, null, null, null);
     when(this.taskModel.getTaskName()).thenReturn(TASK_NAME);
   }
 
diff --git a/samza-core/src/test/java/org/apache/samza/coordinator/TestJobCoordinatorMetadataManager.java b/samza-core/src/test/java/org/apache/samza/coordinator/TestJobCoordinatorMetadataManager.java
index d623aed..b1739d9 100644
--- a/samza-core/src/test/java/org/apache/samza/coordinator/TestJobCoordinatorMetadataManager.java
+++ b/samza-core/src/test/java/org/apache/samza/coordinator/TestJobCoordinatorMetadataManager.java
@@ -141,13 +141,13 @@
     JobCoordinatorMetadata newMetadataWithNoChange =
         new JobCoordinatorMetadata(OLD_EPOCH_ID, OLD_CONFIG_ID, OLD_JOB_MODEL_ID);
     assertEquals("Application attempt count should be 0", 0,
-        metrics.getApplicationAttemptCount().getCount());
+        metrics.getApplicationAttemptCount().getValue().intValue());
 
     metadataChanged =
         jobCoordinatorMetadataManager.checkForMetadataChanges(previousMetadata, newMetadataWithNoChange);
     assertFalse("Metadata check should return false", metadataChanged);
     assertEquals("Application attempt count should be 1", 1,
-        metrics.getApplicationAttemptCount().getCount());
+        metrics.getApplicationAttemptCount().getValue().intValue());
   }
 
   @Test
@@ -161,7 +161,7 @@
     } catch (Exception e) {
       assertTrue("Expecting SamzaException to be thrown", e instanceof SamzaException);
       assertEquals("Metadata generation failed count should be 1", 1,
-          jobCoordinatorMetadataManager.getMetrics().getMetadataGenerationFailedCount().getCount());
+          jobCoordinatorMetadataManager.getMetrics().getMetadataGenerationFailedCount().getValue().intValue());
     }
   }
 
@@ -211,7 +211,7 @@
     JobCoordinatorMetadata actualMetadata = jobCoordinatorMetadataManager.readJobCoordinatorMetadata();
     assertNull("Read failed should return null", actualMetadata);
     assertEquals("Metadata read failed count should be 1", 1,
-        jobCoordinatorMetadataManager.getMetrics().getMetadataReadFailedCount().getCount());
+        jobCoordinatorMetadataManager.getMetrics().getMetadataReadFailedCount().getValue().intValue());
   }
 
   @Test
@@ -240,7 +240,7 @@
     } catch (Exception e) {
       assertTrue("Expecting SamzaException to be thrown", e instanceof SamzaException);
       assertEquals("Metadata write failed count should be 1", 1,
-          jobCoordinatorMetadataManager.getMetrics().getMetadataWriteFailedCount().getCount());
+          jobCoordinatorMetadataManager.getMetrics().getMetadataWriteFailedCount().getValue().intValue());
     }
   }
 }
diff --git a/samza-core/src/test/java/org/apache/samza/job/TestShellCommandBuilder.java b/samza-core/src/test/java/org/apache/samza/job/TestShellCommandBuilder.java
new file mode 100644
index 0000000..ca7be0e
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/job/TestShellCommandBuilder.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.job;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Map;
+import com.google.common.collect.ImmutableMap;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.ShellCommandConfig;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+
+public class TestShellCommandBuilder {
+  private static final String URL_STRING = "http://www.google.com";
+
+  @Test
+  public void testBasicBuild() throws MalformedURLException {
+    Config config = new MapConfig(ImmutableMap.of(ShellCommandConfig.COMMAND_SHELL_EXECUTE, "foo"));
+    ShellCommandBuilder shellCommandBuilder = new ShellCommandBuilder();
+    shellCommandBuilder.setConfig(config);
+    shellCommandBuilder.setId("1");
+    shellCommandBuilder.setUrl(new URL(URL_STRING));
+    Map<String, String> expectedEnvironment = ImmutableMap.of(
+        ShellCommandConfig.ENV_CONTAINER_ID, "1",
+        ShellCommandConfig.ENV_COORDINATOR_URL, URL_STRING,
+        ShellCommandConfig.ENV_JAVA_OPTS, "",
+        ShellCommandConfig.ENV_ADDITIONAL_CLASSPATH_DIR, "");
+    // assertions when command path is not set
+    assertEquals("foo", shellCommandBuilder.buildCommand());
+    assertEquals(expectedEnvironment, shellCommandBuilder.buildEnvironment());
+    // assertions when command path is set to empty string
+    shellCommandBuilder.setCommandPath("");
+    assertEquals("foo", shellCommandBuilder.buildCommand());
+    assertEquals(expectedEnvironment, shellCommandBuilder.buildEnvironment());
+  }
+
+  @Test
+  public void testBuildEnvironment() throws MalformedURLException {
+    Config config = new MapConfig(new ImmutableMap.Builder<String, String>()
+        .put(ShellCommandConfig.COMMAND_SHELL_EXECUTE, "foo")
+        .put(ShellCommandConfig.TASK_JVM_OPTS, "-Xmx4g")
+        .put(ShellCommandConfig.ADDITIONAL_CLASSPATH_DIR, "/path/to/additional/classpath")
+        .put(ShellCommandConfig.TASK_JAVA_HOME, "/path/to/java/home")
+        .build());
+    ShellCommandBuilder shellCommandBuilder = new ShellCommandBuilder();
+    shellCommandBuilder.setConfig(config);
+    shellCommandBuilder.setId("1");
+    shellCommandBuilder.setUrl(new URL(URL_STRING));
+    Map<String, String> expectedEnvironment = new ImmutableMap.Builder<String, String>()
+        .put(ShellCommandConfig.ENV_CONTAINER_ID, "1")
+        .put(ShellCommandConfig.ENV_COORDINATOR_URL, URL_STRING)
+        .put(ShellCommandConfig.ENV_JAVA_OPTS, "-Xmx4g")
+        .put(ShellCommandConfig.ENV_ADDITIONAL_CLASSPATH_DIR, "/path/to/additional/classpath")
+        .put(ShellCommandConfig.ENV_JAVA_HOME, "/path/to/java/home")
+        .build();
+    assertEquals(expectedEnvironment, shellCommandBuilder.buildEnvironment());
+  }
+
+  @Test
+  public void testBuildCommandWithCommandPath() throws MalformedURLException {
+    Config config = new MapConfig(ImmutableMap.of(ShellCommandConfig.COMMAND_SHELL_EXECUTE, "foo"));
+    ShellCommandBuilder shellCommandBuilder = new ShellCommandBuilder();
+    shellCommandBuilder.setConfig(config);
+    shellCommandBuilder.setId("1");
+    shellCommandBuilder.setUrl(new URL(URL_STRING));
+    shellCommandBuilder.setCommandPath("/package/path");
+    assertEquals("/package/path/foo", shellCommandBuilder.buildCommand());
+  }
+}
\ No newline at end of file
diff --git a/samza-core/src/test/java/org/apache/samza/metrics/reporter/TestLoggingMetricsReporter.java b/samza-core/src/test/java/org/apache/samza/metrics/reporter/TestLoggingMetricsReporter.java
new file mode 100644
index 0000000..c1f5972
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/metrics/reporter/TestLoggingMetricsReporter.java
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.metrics.reporter;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import com.google.common.collect.ImmutableMap;
+import org.apache.samza.metrics.Counter;
+import org.apache.samza.metrics.Gauge;
+import org.apache.samza.metrics.Metric;
+import org.apache.samza.metrics.MetricsVisitor;
+import org.apache.samza.metrics.ReadableMetricsRegistry;
+import org.apache.samza.metrics.Snapshot;
+import org.apache.samza.metrics.Timer;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.mockito.stubbing.Answer;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+
+public class TestLoggingMetricsReporter {
+  private static final long LOGGING_INTERVAL_SECONDS = 15;
+  private static final String COUNTER_NAME = "counter_name";
+  private static final long COUNTER_VALUE = 10;
+  private static final String GAUGE_NAME = "gauge_name";
+  private static final double GAUGE_VALUE = 20.0;
+  private static final String TIMER_NAME = "timer_name";
+  private static final double TIMER_VALUE = 30.0;
+  private static final Pattern DEFAULT_PATTERN = Pattern.compile(".*_name");
+  private static final String GROUP_NAME = "group_name";
+  private static final String SOURCE_NAME = "source_name";
+
+  @Mock
+  private ScheduledExecutorService scheduledExecutorService;
+  @Mock
+  private ReadableMetricsRegistry readableMetricsRegistry;
+  @Mock
+  private Counter counter;
+  @Mock
+  private Gauge<Double> gauge;
+  @Mock
+  private Timer timer;
+  @Mock
+  private Snapshot timerSnapshot;
+
+  private LoggingMetricsReporter loggingMetricsReporter;
+
+  @Before
+  public void setup() {
+    MockitoAnnotations.initMocks(this);
+
+    when(this.scheduledExecutorService.scheduleAtFixedRate(any(), eq(LOGGING_INTERVAL_SECONDS),
+        eq(LOGGING_INTERVAL_SECONDS), eq(TimeUnit.SECONDS))).thenAnswer((Answer<Void>) invocation -> {
+          Runnable runnable = invocation.getArgumentAt(0, Runnable.class);
+          runnable.run();
+          return null;
+        });
+
+    when(this.counter.getName()).thenReturn(COUNTER_NAME);
+    when(this.counter.getCount()).thenReturn(COUNTER_VALUE);
+    doAnswer(invocation -> {
+      invocation.getArgumentAt(0, MetricsVisitor.class).counter(this.counter);
+      return null;
+    }).when(this.counter).visit(any());
+
+    when(this.gauge.getName()).thenReturn(GAUGE_NAME);
+    when(this.gauge.getValue()).thenReturn(GAUGE_VALUE);
+    doAnswer(invocation -> {
+      invocation.getArgumentAt(0, MetricsVisitor.class).gauge(this.gauge);
+      return null;
+    }).when(this.gauge).visit(any());
+
+    when(this.timer.getName()).thenReturn(TIMER_NAME);
+    when(this.timer.getSnapshot()).thenReturn(this.timerSnapshot);
+    doAnswer(invocation -> {
+      invocation.getArgumentAt(0, MetricsVisitor.class).timer(this.timer);
+      return null;
+    }).when(this.timer).visit(any());
+    when(this.timerSnapshot.getAverage()).thenReturn(TIMER_VALUE);
+
+    this.loggingMetricsReporter =
+        spy(new LoggingMetricsReporter(this.scheduledExecutorService, DEFAULT_PATTERN, LOGGING_INTERVAL_SECONDS));
+  }
+
+  @Test
+  public void testMetricTypes() {
+    when(this.readableMetricsRegistry.getGroups()).thenReturn(Collections.singleton(GROUP_NAME));
+    Map<String, Metric> metrics =
+        ImmutableMap.of(COUNTER_NAME, this.counter, GAUGE_NAME, this.gauge, TIMER_NAME, this.timer);
+    when(this.readableMetricsRegistry.getGroup(GROUP_NAME)).thenReturn(metrics);
+
+    this.loggingMetricsReporter.register(SOURCE_NAME, this.readableMetricsRegistry);
+    this.loggingMetricsReporter.start();
+
+    verify(this.loggingMetricsReporter).doLog("Metric: source_name-group_name-counter_name, Value: 10");
+    verify(this.loggingMetricsReporter).doLog("Metric: source_name-group_name-gauge_name, Value: 20.0");
+    verify(this.loggingMetricsReporter).doLog("Metric: source_name-group_name-timer_name, Value: 30.0");
+  }
+
+  @Test
+  public void testMultipleRegister() {
+    when(this.readableMetricsRegistry.getGroups()).thenReturn(Collections.singleton(GROUP_NAME));
+    when(this.readableMetricsRegistry.getGroup(GROUP_NAME)).thenReturn(ImmutableMap.of(COUNTER_NAME, this.counter));
+    ReadableMetricsRegistry otherRegistry = mock(ReadableMetricsRegistry.class);
+    String otherGroupName = "other_group";
+    when(otherRegistry.getGroups()).thenReturn(Collections.singleton(otherGroupName));
+    when(otherRegistry.getGroup(otherGroupName)).thenReturn(ImmutableMap.of(GAUGE_NAME, this.gauge));
+
+    this.loggingMetricsReporter.register(SOURCE_NAME, this.readableMetricsRegistry);
+    this.loggingMetricsReporter.register("other_source", otherRegistry);
+    this.loggingMetricsReporter.start();
+
+    verify(this.loggingMetricsReporter).doLog("Metric: source_name-group_name-counter_name, Value: 10");
+    verify(this.loggingMetricsReporter).doLog("Metric: other_source-other_group-gauge_name, Value: 20.0");
+  }
+
+  @Test
+  public void testFiltering() {
+    Pattern countersOnly = Pattern.compile(".*counter.*");
+    this.loggingMetricsReporter =
+        spy(new LoggingMetricsReporter(this.scheduledExecutorService, countersOnly, LOGGING_INTERVAL_SECONDS));
+
+    when(this.readableMetricsRegistry.getGroups()).thenReturn(Collections.singleton(GROUP_NAME));
+    Map<String, Metric> metrics = ImmutableMap.of(COUNTER_NAME, this.counter, GAUGE_NAME, this.gauge);
+    when(this.readableMetricsRegistry.getGroup(GROUP_NAME)).thenReturn(metrics);
+
+    this.loggingMetricsReporter.register(SOURCE_NAME, this.readableMetricsRegistry);
+    this.loggingMetricsReporter.start();
+
+    ArgumentCaptor<String> logs = ArgumentCaptor.forClass(String.class);
+    verify(this.loggingMetricsReporter).doLog(logs.capture());
+    assertEquals(Collections.singletonList("Metric: source_name-group_name-counter_name, Value: 10"),
+        logs.getAllValues());
+  }
+
+  @Test
+  public void testNewMetricsAfterRegister() {
+    when(this.readableMetricsRegistry.getGroups()).thenReturn(Collections.singleton(GROUP_NAME));
+    // first round of logging has one metric (counter only), second call has two (counter and gauge)
+    when(this.readableMetricsRegistry.getGroup(GROUP_NAME)).thenReturn(ImmutableMap.of(COUNTER_NAME, this.counter))
+        .thenReturn(ImmutableMap.of(COUNTER_NAME, this.counter, GAUGE_NAME, this.gauge));
+
+    // capture the logging task so it can be directly executed by the test
+    ArgumentCaptor<Runnable> loggingRunnable = ArgumentCaptor.forClass(Runnable.class);
+    when(this.scheduledExecutorService.scheduleAtFixedRate(loggingRunnable.capture(), eq(LOGGING_INTERVAL_SECONDS),
+        eq(LOGGING_INTERVAL_SECONDS), eq(TimeUnit.SECONDS))).thenReturn(null);
+
+    this.loggingMetricsReporter.register(SOURCE_NAME, this.readableMetricsRegistry);
+    this.loggingMetricsReporter.start();
+
+    // simulate first scheduled execution of logging task
+    loggingRunnable.getValue().run();
+    String expectedCounterLog = "Metric: source_name-group_name-counter_name, Value: 10";
+    // only should get log for counter for the first call
+    verify(this.loggingMetricsReporter).doLog(expectedCounterLog);
+    String expectedGaugeLog = "Metric: source_name-group_name-gauge_name, Value: 20.0";
+    verify(this.loggingMetricsReporter, never()).doLog(expectedGaugeLog);
+
+    // simulate second scheduled execution of logging task
+    loggingRunnable.getValue().run();
+    // should get second log for counter, first log for gauge
+    verify(this.loggingMetricsReporter, times(2)).doLog(expectedCounterLog);
+    verify(this.loggingMetricsReporter).doLog(expectedGaugeLog);
+  }
+}
\ No newline at end of file
diff --git a/samza-core/src/test/java/org/apache/samza/metrics/reporter/TestLoggingMetricsReporterConfig.java b/samza-core/src/test/java/org/apache/samza/metrics/reporter/TestLoggingMetricsReporterConfig.java
new file mode 100644
index 0000000..40b44e7
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/metrics/reporter/TestLoggingMetricsReporterConfig.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.metrics.reporter;
+
+import java.util.Map;
+import com.google.common.collect.ImmutableMap;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.MapConfig;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+
+public class TestLoggingMetricsReporterConfig {
+  private static final String REPORTER_NAME = "reporter_name";
+
+  @Test
+  public void testGetMetricsToLogRegex() {
+    Map<String, String> configMap = ImmutableMap.of("metrics.reporter.reporter_name.log.regex", ".*metric.*");
+    assertEquals(".*metric.*",
+        new LoggingMetricsReporterConfig(new MapConfig(configMap)).getMetricsToLogRegex(REPORTER_NAME));
+  }
+
+  @Test(expected = ConfigException.class)
+  public void testGetMetricsToLogRegexMissing() {
+    new LoggingMetricsReporterConfig(new MapConfig()).getMetricsToLogRegex(REPORTER_NAME);
+  }
+
+  @Test
+  public void testGetLoggingIntervalSeconds() {
+    assertEquals(60, new LoggingMetricsReporterConfig(new MapConfig()).getLoggingIntervalSeconds(REPORTER_NAME));
+
+    Map<String, String> configMap = ImmutableMap.of("metrics.reporter.reporter_name.logging.interval.seconds", "100");
+    assertEquals(100,
+        new LoggingMetricsReporterConfig(new MapConfig(configMap)).getLoggingIntervalSeconds(REPORTER_NAME));
+  }
+}
\ No newline at end of file
diff --git a/samza-core/src/test/java/org/apache/samza/operators/impl/TestWindowOperator.java b/samza-core/src/test/java/org/apache/samza/operators/impl/TestWindowOperator.java
index 76b79a7..8218720 100644
--- a/samza-core/src/test/java/org/apache/samza/operators/impl/TestWindowOperator.java
+++ b/samza-core/src/test/java/org/apache/samza/operators/impl/TestWindowOperator.java
@@ -36,6 +36,7 @@
 import org.apache.samza.container.TaskName;
 import org.apache.samza.context.Context;
 import org.apache.samza.context.MockContext;
+import org.apache.samza.context.TaskContextImpl;
 import org.apache.samza.system.descriptors.GenericInputDescriptor;
 import org.apache.samza.system.descriptors.GenericSystemDescriptor;
 import org.apache.samza.job.model.TaskModel;
@@ -93,11 +94,13 @@
     Serde storeKeySerde = new TimeSeriesKeySerde(new IntegerSerde());
     Serde storeValSerde = KVSerde.of(new IntegerSerde(), new IntegerSerde());
 
+    SystemStreamPartition ssp = new SystemStreamPartition("kafka", "integers", new Partition(0));
     TaskModel taskModel = mock(TaskModel.class);
-    when(taskModel.getSystemStreamPartitions()).thenReturn(ImmutableSet
-        .of(new SystemStreamPartition("kafka", "integers", new Partition(0))));
+    when(taskModel.getSystemStreamPartitions()).thenReturn(ImmutableSet.of(ssp));
     when(taskModel.getTaskName()).thenReturn(new TaskName("task 1"));
     when(this.context.getTaskContext().getTaskModel()).thenReturn(taskModel);
+    when(((TaskContextImpl) this.context.getTaskContext()).getSspsExcludingSideInputs()).thenReturn(
+        ImmutableSet.of(ssp));
     when(this.context.getTaskContext().getTaskMetricsRegistry()).thenReturn(new MetricsRegistryMap());
     when(this.context.getContainerContext().getContainerMetricsRegistry()).thenReturn(new MetricsRegistryMap());
     when(this.context.getTaskContext().getStore("jobName-jobId-window-w1"))
diff --git a/samza-core/src/test/java/org/apache/samza/system/inmemory/TestInMemorySystem.java b/samza-core/src/test/java/org/apache/samza/system/inmemory/TestInMemorySystem.java
index 0a2e221..9439b01 100644
--- a/samza-core/src/test/java/org/apache/samza/system/inmemory/TestInMemorySystem.java
+++ b/samza-core/src/test/java/org/apache/samza/system/inmemory/TestInMemorySystem.java
@@ -127,7 +127,7 @@
   }
 
   @Test
-  public void testEndOfStreamMessage() {
+  public void testEndOfStreamMessageWithTask() {
     EndOfStreamMessage eos = new EndOfStreamMessage("test-task");
 
     produceMessages(eos);
@@ -139,6 +139,24 @@
     List<IncomingMessageEnvelope> results = consumeRawMessages(sspsToPoll);
 
     assertEquals(1, results.size());
+    assertEquals("test-task", ((EndOfStreamMessage) results.get(0).getMessage()).getTaskName());
+    assertFalse(results.get(0).isEndOfStream());
+  }
+
+  @Test
+  public void testEndOfStreamMessageWithoutTask() {
+    EndOfStreamMessage eos = new EndOfStreamMessage();
+
+    produceMessages(eos);
+
+    Set<SystemStreamPartition> sspsToPoll = IntStream.range(0, PARTITION_COUNT)
+        .mapToObj(partition -> new SystemStreamPartition(SYSTEM_STREAM, new Partition(partition)))
+        .collect(Collectors.toSet());
+
+    List<IncomingMessageEnvelope> results = consumeRawMessages(sspsToPoll);
+
+    assertEquals(1, results.size());
+    assertNull(((EndOfStreamMessage) results.get(0).getMessage()).getTaskName());
     assertTrue(results.get(0).isEndOfStream());
   }
 
diff --git a/samza-core/src/test/java/org/apache/samza/util/TestSplitDeploymentUtil.java b/samza-core/src/test/java/org/apache/samza/util/TestSplitDeploymentUtil.java
deleted file mode 100644
index 72772ba..0000000
--- a/samza-core/src/test/java/org/apache/samza/util/TestSplitDeploymentUtil.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.samza.util;
-
-import org.apache.samza.clustermanager.ClusterBasedJobCoordinatorRunner;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.powermock.api.mockito.PowerMockito;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
-
-import static org.junit.Assert.*;
-import static org.mockito.AdditionalMatchers.*;
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.verify;
-import static org.powermock.api.mockito.PowerMockito.*;
-
-
-@RunWith(PowerMockRunner.class)
-@PrepareForTest({ClusterBasedJobCoordinatorRunner.class})
-public class TestSplitDeploymentUtil {
-
-  @Test
-  public void testRunWithIsolatingClassLoader() throws Exception {
-    // partially mock ClusterBasedJobCoordinator (mock runClusterBasedJobCoordinator method only)
-    PowerMockito.spy(ClusterBasedJobCoordinatorRunner.class);
-    // save the context classloader to make sure that it gets set properly once the test is finished
-    ClassLoader previousContextClassLoader = Thread.currentThread().getContextClassLoader();
-    ClassLoader classLoader = mock(ClassLoader.class);
-    String[] args = new String[]{"arg0", "arg1"};
-    doReturn(ClusterBasedJobCoordinatorRunner.class).when(classLoader).loadClass(ClusterBasedJobCoordinatorRunner.class.getName());
-
-    // stub the private static method which is called by reflection
-    PowerMockito.doAnswer(invocation -> {
-        // make sure the only calls to this method has the expected arguments
-      assertArrayEquals(args, invocation.getArgumentAt(0, String[].class));
-      // checks that the context classloader is set correctly
-      assertEquals(classLoader, Thread.currentThread().getContextClassLoader());
-      return null;
-    }).when(ClusterBasedJobCoordinatorRunner.class, "runClusterBasedJobCoordinator", any());
-
-    try {
-      SplitDeploymentUtil.runWithClassLoader(classLoader,
-          ClusterBasedJobCoordinatorRunner.class, "runClusterBasedJobCoordinator", args);
-      assertEquals(previousContextClassLoader, Thread.currentThread().getContextClassLoader());
-    } finally {
-      // reset it explicitly just in case runWithClassLoader throws an exception
-      Thread.currentThread().setContextClassLoader(previousContextClassLoader);
-    }
-    // make sure that the classloader got used
-    verify(classLoader).loadClass(ClusterBasedJobCoordinatorRunner.class.getName());
-    // make sure runClusterBasedJobCoordinator only got called once
-    verifyPrivate(ClusterBasedJobCoordinatorRunner.class).invoke("runClusterBasedJobCoordinator", new Object[]{aryEq(args)});
-  }
-}
diff --git a/samza-core/src/test/scala/org/apache/samza/job/TestShellCommandBuilder.scala b/samza-core/src/test/scala/org/apache/samza/job/TestShellCommandBuilder.scala
deleted file mode 100644
index c70af8d..0000000
--- a/samza-core/src/test/scala/org/apache/samza/job/TestShellCommandBuilder.scala
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.samza.job
-
-import org.junit.Assert._
-import org.junit.Test
-import scala.collection.JavaConverters._
-import org.apache.samza.config.MapConfig
-import org.apache.samza.config.ShellCommandConfig
-import java.net.URL
-
-class TestShellCommandBuilder {
-  @Test
-  def testEnvironmentVariables {
-    val urlStr = "http://www.google.com"
-    val config = new MapConfig(Map(ShellCommandConfig.COMMAND_SHELL_EXECUTE -> "foo").asJava)
-    val scb = new ShellCommandBuilder
-    scb.setConfig(config)
-    scb.setId("1")
-    scb.setUrl(new URL(urlStr))
-    val command = scb.buildCommand
-    val environment = scb.buildEnvironment
-    assertEquals("foo", command)
-    assertEquals("1", environment.get(ShellCommandConfig.ENV_CONTAINER_ID))
-    assertEquals(urlStr, environment.get(ShellCommandConfig.ENV_COORDINATOR_URL))
-  }
-
-  // if cmdPath is specified, the full path to the command should be adjusted
-  @Test
-  def testBuildCommandWithCommandPath {
-    val urlStr = "http://www.linkedin.com"
-    val config = new MapConfig(Map(ShellCommandConfig.COMMAND_SHELL_EXECUTE -> "foo").asJava)
-    val scb = new ShellCommandBuilder
-    scb.setConfig(config)
-    scb.setId("1")
-    scb.setUrl(new URL(urlStr))
-    val command = scb.buildCommand
-    assertEquals("foo", command)
-
-    scb.setCommandPath("/package/path")
-    val command1 = scb.buildCommand
-    assertEquals("/package/path/foo", command1)
-  }
-}
\ No newline at end of file
diff --git a/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.java b/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.java
index 01c22f5..97443e3 100644
--- a/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.java
+++ b/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.java
@@ -55,10 +55,6 @@
     Preconditions.checkNotNull(type);
     Preconditions.checkState(!grouperFactoryClassName.isEmpty(), "Empty grouper factory class provided");
 
-    Preconditions.checkState(CHECKPOINT_V1_KEY_TYPE.equals(type) || CHECKPOINT_V2_KEY_TYPE.equals(type),
-        String.format("Invalid type provided for checkpoint key. Expected: (%s or %s) Actual: (%s)",
-            CHECKPOINT_V1_KEY_TYPE, CHECKPOINT_V2_KEY_TYPE, type));
-
     this.grouperFactoryClassName = grouperFactoryClassName;
     this.taskName = taskName;
     this.type = type;
diff --git a/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKeySerde.java b/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKeySerde.java
index c5c431b..59ff34f 100644
--- a/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKeySerde.java
+++ b/samza-kafka/src/main/java/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKeySerde.java
@@ -57,9 +57,8 @@
   public KafkaCheckpointLogKey fromBytes(byte[] bytes) {
     try {
       LinkedHashMap<String, String> deserializedKey = MAPPER.readValue(bytes, LinkedHashMap.class);
-      String key = deserializedKey.get(TYPE_FIELD);
 
-      return new KafkaCheckpointLogKey(key, new TaskName(deserializedKey.get(TASK_NAME_FIELD)),
+      return new KafkaCheckpointLogKey(deserializedKey.get(TYPE_FIELD), new TaskName(deserializedKey.get(TASK_NAME_FIELD)),
           deserializedKey.get(SSP_GROUPER_FACTORY_FIELD));
     } catch (Exception e) {
       throw new SamzaException(String.format("Exception in de-serializing checkpoint bytes: %s",
diff --git a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
index e719439..7dbb9b3 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
@@ -335,7 +335,7 @@
     partitionMetaData.getOldestOffset
   }
 
-  private def buildOutgoingMessageEnvelope[T <: Checkpoint](taskName: TaskName, checkpoint: T): OutgoingMessageEnvelope = {
+  def buildOutgoingMessageEnvelope[T <: Checkpoint](taskName: TaskName, checkpoint: T): OutgoingMessageEnvelope = {
     checkpoint match {
       case checkpointV1: CheckpointV1 => {
         val key = new KafkaCheckpointLogKey(
diff --git a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
index 69a9966..391536a 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
@@ -381,14 +381,12 @@
   }
 
   def getKafkaSystemProducerConfig( systemName: String,
-                                    clientId: String,
-                                    injectedProps: Map[String, String] = Map()) = {
+                                    clientId: String) = {
 
     val subConf = config.subset("systems.%s.producer." format systemName, true)
     val producerProps = new util.HashMap[String, String]()
     producerProps.putAll(subConf)
     producerProps.put("client.id", clientId)
-    producerProps.putAll(injectedProps.asJava)
     new KafkaProducerConfig(systemName, clientId, producerProps)
   }
 }
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
index 8d1fd6b..a2773f6 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
@@ -32,14 +32,6 @@
 import org.apache.samza.util._
 
 object KafkaSystemFactory extends Logging {
-  @VisibleForTesting
-  def getInjectedProducerProperties(systemName: String, config: Config) = if (new StorageConfig(config).isChangelogSystem(systemName)) {
-    warn("System name '%s' is being used as a changelog. Disabling compression since Kafka does not support compression for log compacted topics." format systemName)
-    Map[String, String]("compression.type" -> "none")
-  } else {
-    Map[String, String]()
-  }
-
   val CLIENTID_PRODUCER_PREFIX = "kafka-producer"
   val CLIENTID_CONSUMER_PREFIX = "kafka-consumer"
   val CLIENTID_ADMIN_PREFIX = "kafka-admin-consumer"
@@ -67,9 +59,8 @@
   }
 
   def getProducer(systemName: String, config: Config, registry: MetricsRegistry): SystemProducer = {
-    val injectedProps = KafkaSystemFactory.getInjectedProducerProperties(systemName, config)
     val clientId = KafkaConsumerConfig.createClientId(KafkaSystemFactory.CLIENTID_PRODUCER_PREFIX, config);
-    val producerConfig = config.getKafkaSystemProducerConfig(systemName, clientId, injectedProps)
+    val producerConfig = config.getKafkaSystemProducerConfig(systemName, clientId)
     val getProducer = () => {
       new KafkaProducer[Array[Byte], Array[Byte]](producerConfig.getProducerProperties)
     }
diff --git a/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKeySerde.java b/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKeySerde.java
index 7245e70..08fa02c 100644
--- a/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKeySerde.java
+++ b/samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointLogKeySerde.java
@@ -53,11 +53,23 @@
 
   @Test
   public void testCheckpointTypeV2() {
-    KafkaCheckpointLogKey keyV2 = new KafkaCheckpointLogKey(KafkaCheckpointLogKey.CHECKPOINT_V2_KEY_TYPE,
-        new TaskName("Partition 0"), GroupByPartitionFactory.class.getCanonicalName());
+    KafkaCheckpointLogKey keyV2 = new KafkaCheckpointLogKey(KafkaCheckpointLogKey.CHECKPOINT_V2_KEY_TYPE, new TaskName("Partition 0"),
+        GroupByPartitionFactory.class.getCanonicalName());
     KafkaCheckpointLogKeySerde checkpointKeySerde = new KafkaCheckpointLogKeySerde();
 
     // test that deserialize(serialize(k)) == k
     Assert.assertEquals(keyV2, checkpointKeySerde.fromBytes(checkpointKeySerde.toBytes(keyV2)));
   }
+
+  @Test
+  public void testForwardsCompatibility() {
+    // Set the key to another value, this is for the future if we want to support multiple checkpoint keys
+    // we do not want to throw in the Serdes layer, but must be validated in the CheckpointManager
+    KafkaCheckpointLogKey key = new KafkaCheckpointLogKey("checkpoint-v2",
+        new TaskName("Partition 0"), GroupByPartitionFactory.class.getCanonicalName());
+    KafkaCheckpointLogKeySerde checkpointSerde = new KafkaCheckpointLogKeySerde();
+
+    // test that deserialize(serialize(k)) == k
+    Assert.assertEquals(key, checkpointSerde.fromBytes(checkpointSerde.toBytes(key)));
+  }
 }
diff --git a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
index 866904b..835f53e 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
@@ -20,7 +20,6 @@
 package org.apache.samza.checkpoint.kafka
 
 import java.util.Properties
-
 import kafka.integration.KafkaServerTestHarness
 import kafka.utils.{CoreUtils, TestUtils}
 import com.google.common.collect.ImmutableMap
@@ -29,7 +28,7 @@
 import org.apache.samza.container.TaskName
 import org.apache.samza.container.grouper.stream.GroupByPartitionFactory
 import org.apache.samza.metrics.MetricsRegistry
-import org.apache.samza.serializers.CheckpointV1Serde
+import org.apache.samza.serializers.{CheckpointV1Serde, CheckpointV2Serde}
 import org.apache.samza.system._
 import org.apache.samza.system.kafka.{KafkaStreamSpec, KafkaSystemFactory}
 import org.apache.samza.util.ScalaJavaUtil.JavaOptionals
@@ -276,6 +275,42 @@
   }
 
   @Test
+  def testReadCheckpointShouldIgnoreUnknownCheckpointKeys(): Unit = {
+      val checkpointTopic = "checkpoint-topic-1"
+      val kcm1 = createKafkaCheckpointManager(checkpointTopic)
+      kcm1.register(taskName)
+      kcm1.createResources
+      kcm1.start
+      kcm1.stop
+
+      // check that start actually creates the topic with log compaction enabled
+      val topicConfig = adminZkClient.getAllTopicConfigs().getOrElse(checkpointTopic, new Properties())
+
+      assertEquals(topicConfig, new KafkaConfig(config).getCheckpointTopicProperties())
+      assertEquals("compact", topicConfig.get("cleanup.policy"))
+      assertEquals("26214400", topicConfig.get("segment.bytes"))
+
+      // read before topic exists should result in a null checkpoint
+      val readCp = readCheckpoint(checkpointTopic, taskName)
+      assertNull(readCp)
+    // skips unknown checkpoints from checkpoint topic
+    writeCheckpoint(checkpointTopic, taskName, checkpoint1, "checkpoint-v2", useMock = true)
+    assertNull(readCheckpoint(checkpointTopic, taskName, useMock = true))
+
+    // reads latest v1 checkpoints
+    writeCheckpoint(checkpointTopic, taskName, checkpoint1, useMock = true)
+    assertEquals(checkpoint1, readCheckpoint(checkpointTopic, taskName, useMock = true))
+
+    // writing checkpoint v2 still returns the previous v1 checkpoint
+    writeCheckpoint(checkpointTopic, taskName, checkpoint2, "checkpoint-v2", useMock = true)
+    assertEquals(checkpoint1, readCheckpoint(checkpointTopic, taskName, useMock = true))
+
+    // writing checkpoint2 with the correct key returns the checkpoint2
+    writeCheckpoint(checkpointTopic, taskName, checkpoint2, useMock = true)
+    assertEquals(checkpoint2, readCheckpoint(checkpointTopic, taskName, useMock = true))
+  }
+
+  @Test
   def testWriteCheckpointShouldRetryFiniteTimesOnFailure(): Unit = {
     val checkpointTopic = "checkpoint-topic-2"
     val mockKafkaProducer: SystemProducer = Mockito.mock(classOf[SystemProducer])
@@ -401,7 +436,8 @@
   }
 
   private def createKafkaCheckpointManager(cpTopic: String, serde: CheckpointV1Serde = new CheckpointV1Serde,
-    failOnTopicValidation: Boolean = true, overrideConfig: Config = config) = {
+    failOnTopicValidation: Boolean = true, useMock: Boolean = false, checkpointKey: String = KafkaCheckpointLogKey.CHECKPOINT_V1_KEY_TYPE,
+    overrideConfig: Config = config) = {
     val kafkaConfig = new org.apache.samza.config.KafkaConfig(overrideConfig)
     val props = kafkaConfig.getCheckpointTopicProperties()
     val systemName = kafkaConfig.getCheckpointSystem.getOrElse(
@@ -414,11 +450,17 @@
     val systemFactory = ReflectionUtil.getObj(systemFactoryClassName, classOf[SystemFactory])
 
     val spec = new KafkaStreamSpec("id", cpTopic, checkpointSystemName, 1, 1, props)
-    new KafkaCheckpointManager(spec, systemFactory, failOnTopicValidation, overrideConfig, new NoOpMetricsRegistry, serde)
+
+    if (useMock) {
+      new MockKafkaCheckpointManager(spec, systemFactory, failOnTopicValidation, serde, checkpointKey)
+    } else {
+      new KafkaCheckpointManager(spec, systemFactory, failOnTopicValidation, overrideConfig, new NoOpMetricsRegistry, serde)
+    }
   }
 
-  private def readCheckpoint(checkpointTopic: String, taskName: TaskName, config: Config = config) : Checkpoint = {
-    val kcm = createKafkaCheckpointManager(checkpointTopic, overrideConfig = config)
+  private def readCheckpoint(checkpointTopic: String, taskName: TaskName, config: Config = config,
+    useMock: Boolean = false) : Checkpoint = {
+    val kcm = createKafkaCheckpointManager(checkpointTopic, overrideConfig = config, useMock = useMock)
     kcm.register(taskName)
     kcm.start
     val checkpoint = kcm.readLastCheckpoint(taskName)
@@ -426,8 +468,9 @@
     checkpoint
   }
 
-  private def writeCheckpoint(checkpointTopic: String, taskName: TaskName, checkpoint: Checkpoint): Unit = {
-    val kcm = createKafkaCheckpointManager(checkpointTopic)
+  private def writeCheckpoint(checkpointTopic: String, taskName: TaskName, checkpoint: Checkpoint,
+    checkpointKey: String = KafkaCheckpointLogKey.CHECKPOINT_V1_KEY_TYPE, useMock: Boolean = false): Unit = {
+    val kcm = createKafkaCheckpointManager(checkpointTopic, checkpointKey = checkpointKey, useMock = useMock)
     kcm.register(taskName)
     kcm.start
     kcm.writeCheckpoint(taskName, checkpoint)
@@ -456,4 +499,35 @@
     }
   }
 
+
+  class MockKafkaCheckpointManager(spec: KafkaStreamSpec, systemFactory: SystemFactory, failOnTopicValidation: Boolean,
+    serde: CheckpointV1Serde = new CheckpointV1Serde, checkpointKey: String)
+    extends KafkaCheckpointManager(spec, systemFactory, failOnTopicValidation, config,
+      new NoOpMetricsRegistry, serde) {
+
+    override def buildOutgoingMessageEnvelope[T <: Checkpoint](taskName: TaskName, checkpoint: T): OutgoingMessageEnvelope = {
+      val key = new KafkaCheckpointLogKey(checkpointKey, taskName, expectedGrouperFactory)
+      val keySerde = new KafkaCheckpointLogKeySerde
+      val checkpointMsgSerde = new CheckpointV1Serde
+      val checkpointV2MsgSerde = new CheckpointV2Serde
+      val keyBytes = try {
+        keySerde.toBytes(key)
+      } catch {
+        case e: Exception => throw new SamzaException(s"Exception when writing checkpoint-key for $taskName: $checkpoint", e)
+      }
+      val msgBytes = try {
+        checkpoint match {
+          case v1: CheckpointV1 =>
+            checkpointMsgSerde.toBytes(v1)
+          case v2: CheckpointV2 =>
+            checkpointV2MsgSerde.toBytes(v2)
+          case _ =>
+            throw new IllegalArgumentException("Unknown checkpoint key type for test, please use Checkpoint v1 or v2")
+        }
+      } catch {
+        case e: Exception => throw new SamzaException(s"Exception when writing checkpoint for $taskName: $checkpoint", e)
+      }
+      new OutgoingMessageEnvelope(checkpointSsp, keyBytes, msgBytes)
+    }
+  }
 }
diff --git a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemFactory.scala b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemFactory.scala
index 596d67b..ecbc00d 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemFactory.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemFactory.scala
@@ -82,16 +82,4 @@
     assertNotNull(producer)
     assertTrue(producer.isInstanceOf[KafkaSystemProducer])
   }
-
-  @Test
-  def testInjectedProducerProps {
-    val configMap = Map[String, String](
-      StorageConfig.FACTORY.format("system1") -> "some.factory.Class",
-      StorageConfig.CHANGELOG_STREAM.format("system1") -> "system1.stream1",
-      StorageConfig.FACTORY.format("system2") -> "some.factory.Class")
-    val config = new MapConfig(configMap.asJava)
-    assertEquals(Map[String, String](), KafkaSystemFactory.getInjectedProducerProperties("system3", config))
-    assertEquals(Map[String, String](), KafkaSystemFactory.getInjectedProducerProperties("system2", config))
-    assertEquals(Map[String, String]("compression.type" -> "none"), KafkaSystemFactory.getInjectedProducerProperties("system1", config))
-  }
 }
diff --git a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbOptionsHelper.java b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbOptionsHelper.java
index 0908a97..ac2b3f0 100644
--- a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbOptionsHelper.java
+++ b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbOptionsHelper.java
@@ -69,6 +69,8 @@
   private static final String ROCKSDB_KEEP_LOG_FILE_NUM = "rocksdb.keep.log.file.num";
   private static final String ROCKSDB_DELETE_OBSOLETE_FILES_PERIOD_MICROS = "rocksdb.delete.obsolete.files.period.micros";
   private static final String ROCKSDB_MAX_MANIFEST_FILE_SIZE = "rocksdb.max.manifest.file.size";
+  private static final String ROCKSDB_MAX_OPEN_FILES = "rocksdb.max.open.files";
+  private static final String ROCKSDB_MAX_FILE_OPENING_THREADS = "rocksdb.max.file.opening.threads";
 
   public static Options options(Config storeConfig, int numTasksForContainer, File storeDir, StorageEngineFactory.StoreMode storeMode) {
     Options options = new Options();
@@ -124,6 +126,8 @@
     options.setMaxLogFileSize(storeConfig.getLong(ROCKSDB_MAX_LOG_FILE_SIZE_BYTES, 64 * 1024 * 1024L));
     options.setKeepLogFileNum(storeConfig.getLong(ROCKSDB_KEEP_LOG_FILE_NUM, 2));
     options.setDeleteObsoleteFilesPeriodMicros(storeConfig.getLong(ROCKSDB_DELETE_OBSOLETE_FILES_PERIOD_MICROS, 21600000000L));
+    options.setMaxOpenFiles(storeConfig.getInt(ROCKSDB_MAX_OPEN_FILES, -1));
+    options.setMaxFileOpeningThreads(storeConfig.getInt(ROCKSDB_MAX_FILE_OPENING_THREADS, 16));
     // The default for rocksdb is 18446744073709551615, which is larger than java Long.MAX_VALUE. Hence setting it only if it's passed.
     if (storeConfig.containsKey(ROCKSDB_MAX_MANIFEST_FILE_SIZE)) {
       options.setMaxManifestFileSize(storeConfig.getLong(ROCKSDB_MAX_MANIFEST_FILE_SIZE));
diff --git a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java
index 0044904..f9a1f24 100644
--- a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java
+++ b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java
@@ -49,6 +49,9 @@
   static final public String ROCKSDB_NUM_WRITE_BUFFERS = "rocksdb.num.write.buffers";
   static final public String ROCKSDB_MAX_LOG_FILE_SIZE_BYTES = "rocksdb.max.log.file.size.bytes";
   static final public String ROCKSDB_KEEP_LOG_FILE_NUM = "rocksdb.keep.log.file.num";
+  static final public String ROCKSDB_MAX_OPEN_FILES = "rocksdb.max.open.files";
+  static final public String ROCKSDB_MAX_FILE_OPENING_THREADS = "rocksdb.max.file.opening.threads";
+
 
   private Integer writeBatchSize;
   private Integer objectCacheSize;
@@ -61,6 +64,8 @@
   private Integer numLogFilesToKeep;
   private String compressionType;
   private String compactionStyle;
+  private Integer maxOpenFiles;
+  private Integer maxFileOpeningThreads;
 
   /**
    * Constructs a table descriptor instance
@@ -273,6 +278,35 @@
     return this;
   }
 
+  /**
+   * Limits the number of open files that can be used by the DB.  You may need to increase this if your database
+   * has a large working set. Value -1 means files opened are always kept open.
+   * <p>
+   *  Default value is -1.
+   * <p>
+   * @param maxOpenFiles the number of open files that can be used by the DB.
+   * @return this table descriptor instance
+   */
+  public RocksDbTableDescriptor<K, V> withMaxOpenFiles(int maxOpenFiles) {
+    this.maxOpenFiles = maxOpenFiles;
+    return this;
+  }
+
+  /**
+   * Sets the number of threads used to open DB files.
+   * If max_open_files is -1, DB will open all files on DB::Open(). You can use this option to increase the number of
+   * threads used to open files.
+   * <p>
+   * Default is 16.
+   * <p>
+   * @param maxFileOpeningThreads The number of threads to use when opening DB files.
+   * @return the table descriptor instance
+   */
+  public RocksDbTableDescriptor<K, V> withMaxFileOpeningThreads(int maxFileOpeningThreads) {
+    this.maxFileOpeningThreads = maxFileOpeningThreads;
+    return this;
+  }
+
   @Override
   public String getProviderFactoryClassName() {
     return LocalTableProviderFactory.class.getName();
@@ -320,6 +354,12 @@
     if (numLogFilesToKeep != null) {
       addStoreConfig(ROCKSDB_KEEP_LOG_FILE_NUM, numLogFilesToKeep.toString(), tableConfig);
     }
+    if (maxOpenFiles != null) {
+      addStoreConfig(ROCKSDB_MAX_OPEN_FILES, maxOpenFiles.toString(), tableConfig);
+    }
+    if (maxFileOpeningThreads != null) {
+      addStoreConfig(ROCKSDB_MAX_FILE_OPENING_THREADS, maxFileOpeningThreads.toString(), tableConfig);
+    }
 
     return Collections.unmodifiableMap(tableConfig);
   }
diff --git a/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java b/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java
index c58d123..9fc9938 100644
--- a/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java
+++ b/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java
@@ -73,10 +73,12 @@
         .withTtl(7)
         .withWriteBatchSize(8)
         .withWriteBufferSize(9)
+        .withMaxOpenFiles(10)
+        .withMaxFileOpeningThreads(11)
         .withConfig("abc", "xyz")
         .toConfig(createJobConfig());
 
-    Assert.assertEquals(14, tableConfig.size());
+    Assert.assertEquals(16, tableConfig.size());
     assertEquals("1", RocksDbTableDescriptor.ROCKSDB_BLOCK_SIZE_BYTES, tableConfig);
     assertEquals("2", RocksDbTableDescriptor.CONTAINER_CACHE_SIZE_BYTES, tableConfig);
     assertEquals("3", RocksDbTableDescriptor.ROCKSDB_MAX_LOG_FILE_SIZE_BYTES, tableConfig);
@@ -86,6 +88,8 @@
     assertEquals("7", RocksDbTableDescriptor.ROCKSDB_TTL_MS, tableConfig);
     assertEquals("8", RocksDbTableDescriptor.WRITE_BATCH_SIZE, tableConfig);
     assertEquals("9", RocksDbTableDescriptor.CONTAINER_WRITE_BUFFER_SIZE_BYTES, tableConfig);
+    assertEquals("10", RocksDbTableDescriptor.ROCKSDB_MAX_OPEN_FILES, tableConfig);
+    assertEquals("11", RocksDbTableDescriptor.ROCKSDB_MAX_FILE_OPENING_THREADS, tableConfig);
     assertEquals("snappy", RocksDbTableDescriptor.ROCKSDB_COMPRESSION, tableConfig);
     assertEquals("fifo", RocksDbTableDescriptor.ROCKSDB_COMPACTION_STYLE, tableConfig);
     Assert.assertFalse(tableConfig.containsKey(String.format(StorageConfig.CHANGELOG_STREAM, TABLE_ID)));
diff --git a/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java b/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
index 2b03c31..f82a576 100644
--- a/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
+++ b/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
@@ -81,7 +81,7 @@
   private static final String CREATE_STREAM_ENABLED = "task.log4j.create.stream.enabled";
 
   private static final long DEFAULT_QUEUE_TIMEOUT_S = 2; // Abitrary choice
-  private final BlockingQueue<byte[]> logQueue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_SIZE);
+  private final BlockingQueue<EncodedLogEvent> logQueue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_SIZE);
 
   private SystemStream systemStream = null;
   private SystemProducer systemProducer = null;
@@ -233,13 +233,13 @@
    */
   private void handleEvent(LogEvent event) throws InterruptedException {
     if (usingAsyncLogger) {
-      sendEventToSystemProducer(encodeLogEventToBytes(event));
+      sendEventToSystemProducer(encodeLogEvent(event));
       return;
     }
 
     // Serialize the event before adding to the queue to leverage the caller thread
     // and ensure that the transferThread can keep up.
-    if (!logQueue.offer(encodeLogEventToBytes(event), queueTimeoutS, TimeUnit.SECONDS)) {
+    if (!logQueue.offer(encodeLogEvent(event), queueTimeoutS, TimeUnit.SECONDS)) {
       // Do NOT retry adding system to the queue. Dropping the event allows us to alleviate the unlikely
       // possibility of a deadlock, which can arise due to a circular dependency between the SystemProducer
       // which is used for StreamAppender and the log, which uses StreamAppender. Any locks held in the callstack
@@ -265,8 +265,8 @@
     metrics.bufferFillPct.set(Math.round(100f * logQueue.size() / DEFAULT_QUEUE_SIZE));
   }
 
-  protected byte[] encodeLogEventToBytes(LogEvent event) {
-    return serde.toBytes(subLog(event));
+  protected EncodedLogEvent encodeLogEvent(LogEvent event) {
+    return new ByteArrayEncodedLogEvent(serde.toBytes(subLog(event)));
   }
 
   private Message subAppend(LogEvent event) {
@@ -437,21 +437,21 @@
 
   /**
    * Helper method to send a serialized log-event to the systemProducer, and increment respective methods.
-   * @param serializedLogEvent
+   * @param logQueueEntry the serialized log-event to be sent to the systemProducer
    */
-  private void sendEventToSystemProducer(byte[] serializedLogEvent) {
-    metrics.logMessagesBytesSent.inc(serializedLogEvent.length);
+  private void sendEventToSystemProducer(EncodedLogEvent logQueueEntry) {
+    metrics.logMessagesBytesSent.inc(logQueueEntry.getEntryValueSize());
     metrics.logMessagesCountSent.inc();
-    systemProducer.send(SOURCE, decorateLogEvent(serializedLogEvent));
+    systemProducer.send(SOURCE, decorateLogEvent(logQueueEntry));
   }
 
   /**
    * Helper method to create an OutgoingMessageEnvelope from the serialized log event.
-   * @param messageBytes message bytes
+   * @param logQueueEntry message bytes
    * @return OutgoingMessageEnvelope that contains the message bytes along with the system stream
    */
-  protected OutgoingMessageEnvelope decorateLogEvent(byte[] messageBytes) {
-    return new OutgoingMessageEnvelope(systemStream, keyBytes, messageBytes);
+  protected OutgoingMessageEnvelope decorateLogEvent(EncodedLogEvent logQueueEntry) {
+    return new OutgoingMessageEnvelope(systemStream, keyBytes, logQueueEntry.getValue());
   }
 
   protected String getStreamName(String jobName, String jobId) {
@@ -498,4 +498,42 @@
   public Serde<LogEvent> getSerde() {
     return serde;
   }
+
+  /**
+   * A LogQeueEntry is the element inserted into the log queue of the stream appender
+   * that holds the log messages before they are sent to the underlying system producer.
+   * @param <T> type of object held as the entry
+   */
+  protected interface EncodedLogEvent<T> {
+    /**
+     * fetches the size of the log message held within the LogQueueEntry
+     * @return size of the log message
+     */
+    public long getEntryValueSize();
+
+    /**
+     * fetches the actual log message held within the LogQueueEntry
+     * @return the actual log message
+     */
+    public T getValue();
+  }
+
+  /**
+   * LogQueueEntry impl that holds the serialized byte[] of the log message.
+   */
+  private class ByteArrayEncodedLogEvent implements EncodedLogEvent<byte[]> {
+    final byte[] entryValue;
+    public ByteArrayEncodedLogEvent(byte[] array) {
+      entryValue = array;
+    }
+    @Override
+    public byte[] getValue() {
+      return entryValue;
+    }
+
+    @Override
+    public long getEntryValueSize() {
+      return entryValue.length;
+    }
+  }
 }
\ No newline at end of file
diff --git a/samza-shell/src/main/bash/run-class.sh b/samza-shell/src/main/bash/run-class.sh
index 9b5ac30..a162ad5 100755
--- a/samza-shell/src/main/bash/run-class.sh
+++ b/samza-shell/src/main/bash/run-class.sh
@@ -53,13 +53,13 @@
 echo APPLICATION_LIB_DIR=$APPLICATION_LIB_DIR
 echo BASE_LIB_DIR=$BASE_LIB_DIR
 
-BASE_LIB_CLASSPATH=""
+CLASSPATH=""
 # all the jars need to be appended on newlines to ensure line argument length of 72 bytes is not violated
 for file in $BASE_LIB_DIR/*.[jw]ar;
 do
-  BASE_LIB_CLASSPATH=$BASE_LIB_CLASSPATH" $file \n"
+  CLASSPATH=$CLASSPATH" $file \n"
 done
-echo generated from BASE_LIB_DIR BASE_LIB_CLASSPATH=$BASE_LIB_CLASSPATH
+echo generated from BASE_LIB_DIR CLASSPATH=$CLASSPATH
 
 # In some cases (AWS) $JAVA_HOME/bin doesn't contain jar.
 if [ -z "$JAVA_HOME" ] || [ ! -e "$JAVA_HOME/bin/jar" ]; then
@@ -68,23 +68,21 @@
   JAR="$JAVA_HOME/bin/jar"
 fi
 
-# Create a pathing JAR for the JARs in the BASE_LIB_DIR
-# Newlines and spaces are intended to ensure proper parsing of manifest in pathing jar
-printf "Class-Path: \n $BASE_LIB_CLASSPATH \n" > base-lib-manifest.txt
-# Creates a new archive and adds custom manifest information to base-lib-pathing.jar
-eval "$JAR -cvmf base-lib-manifest.txt base-lib-pathing.jar"
+# Create a separate directory for writing files related to classpath management. It is easier to manage
+# permissions for the classpath-related files when they are in their own directory. An example of where
+# this is helpful is when using container images which might have predefined permissions for certain
+# directories.
+CLASSPATH_WORKSPACE_DIR=$base_dir/classpath_workspace
+mkdir -p $CLASSPATH_WORKSPACE_DIR
+# file containing the classpath string; used to avoid passing long classpaths directly to the jar command
+PATHING_MANIFEST_FILE=$CLASSPATH_WORKSPACE_DIR/manifest.txt
+# jar file to include on the classpath for running the main class
+PATHING_JAR_FILE=$CLASSPATH_WORKSPACE_DIR/pathing.jar
 
-# Create a pathing JAR for the runtime framework resources. It is useful to separate this from the base-lib-pathing.jar
-# because the split deployment framework may only need the resources from this runtime pathing JAR.
-if ! [[ $HADOOP_CONF_DIR =~ .*/$ ]]; then
-  # manifest requires a directory to have a trailing slash
-  HADOOP_CONF_DIR="$HADOOP_CONF_DIR/"
-fi
-# HADOOP_CONF_DIR should be supplied to classpath explicitly for Yarn to parse configs
-RUNTIME_FRAMEWORK_RESOURCES_CLASSPATH="$HADOOP_CONF_DIR \n"
-# TODO add JARs from ADDITIONAL_CLASSPATH_DIR to runtime-framework-resources-pathing.jar as well
-printf "Class-Path: \n $RUNTIME_FRAMEWORK_RESOURCES_CLASSPATH \n" > runtime-framework-resources-manifest.txt
-eval "$JAR -cvmf runtime-framework-resources-manifest.txt runtime-framework-resources-pathing.jar"
+# Newlines and spaces are intended to ensure proper parsing of manifest in pathing jar
+printf "Class-Path: \n $CLASSPATH \n" > $PATHING_MANIFEST_FILE
+# Creates a new archive and adds custom manifest information to pathing.jar
+eval "$JAR -cvmf $PATHING_MANIFEST_FILE $PATHING_JAR_FILE"
 
 if [ -z "$JAVA_HOME" ]; then
   JAVA="java"
@@ -163,11 +161,12 @@
 # Check if 64 bit is set. If not - try and set it if it's supported
 [[ $JAVA_OPTS != *-d64* ]] && check_and_enable_64_bit_mode
 
-echo $JAVA $JAVA_OPTS -cp base-lib-pathing.jar:runtime-framework-resources-pathing.jar "$@"
+# HADOOP_CONF_DIR should be supplied to classpath explicitly for Yarn to parse configs
+echo $JAVA $JAVA_OPTS -cp $HADOOP_CONF_DIR:$PATHING_JAR_FILE "$@"
 
 ## If localized resource lib directory is defined, then include it in the classpath.
 if [[ -z "${ADDITIONAL_CLASSPATH_DIR}" ]]; then
-  exec $JAVA $JAVA_OPTS -cp base-lib-pathing.jar:runtime-framework-resources-pathing.jar "$@"
+  exec $JAVA $JAVA_OPTS -cp $HADOOP_CONF_DIR:$PATHING_JAR_FILE "$@"
 else
-  exec $JAVA $JAVA_OPTS -cp base-lib-pathing.jar:runtime-framework-resources-pathing.jar:$ADDITIONAL_CLASSPATH_DIR "$@"
-fi
\ No newline at end of file
+  exec $JAVA $JAVA_OPTS -cp $HADOOP_CONF_DIR:$PATHING_JAR_FILE:$ADDITIONAL_CLASSPATH_DIR "$@"
+fi
diff --git a/samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java b/samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java
index fb65eea..943a8ca 100644
--- a/samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java
+++ b/samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java
@@ -107,6 +107,14 @@
     this.configs = new HashMap<>();
     this.inMemoryScope = RandomStringUtils.random(10, true, true);
     configs.put(ApplicationConfig.APP_NAME, APP_NAME);
+    /*
+     * Use a unique app id to help make sure a test execution is isolated from others.
+     * A concrete example of where this helps is to avoid an issue with ControlMessageSender. It has a static cache
+     * keyed by stream id to save partition counts for intermediate streams. This means that different tests can
+     * collide in this cache if they use the same intermediate stream names. Having a unique app id makes the
+     * intermediate streams unique across tests.
+     */
+    configs.put(ApplicationConfig.APP_ID, this.inMemoryScope);
     configs.put(JobConfig.PROCESSOR_ID, "1");
     configs.put(JobCoordinatorConfig.JOB_COORDINATOR_FACTORY, PassthroughJobCoordinatorFactory.class.getName());
     configs.put(JobConfig.STARTPOINT_METADATA_STORE_FACTORY, InMemoryMetadataStoreFactory.class.getCanonicalName());
diff --git a/samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTestHarness.java b/samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTestHarness.java
index 0d1e49a..012eece 100644
--- a/samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTestHarness.java
+++ b/samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTestHarness.java
@@ -188,6 +188,7 @@
     configMap.put("systems.kafka.samza.key.serde", "string");
     configMap.put("systems.kafka.samza.msg.serde", "string");
     configMap.put("systems.kafka.samza.offset.default", "oldest");
+    configMap.put("systems.kafka.producer.compression.type", "snappy");
     configMap.put("job.coordinator.system", "kafka");
     configMap.put("job.default.system", "kafka");
     configMap.put("job.coordinator.replication.factor", "1");
diff --git a/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableEndToEnd.java b/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableEndToEnd.java
index 78fc7b5..ea81f13 100644
--- a/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableEndToEnd.java
+++ b/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableEndToEnd.java
@@ -19,6 +19,7 @@
 
 package org.apache.samza.test.table;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -27,10 +28,8 @@
 
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
-import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.JobCoordinatorConfig;
-import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.TaskConfig;
 import org.apache.samza.container.grouper.task.SingleContainerGrouperFactory;
 import org.apache.samza.context.Context;
@@ -39,7 +38,6 @@
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
 import org.apache.samza.operators.functions.MapFunction;
-import org.apache.samza.runtime.LocalApplicationRunner;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.NoOpSerde;
@@ -47,11 +45,10 @@
 import org.apache.samza.storage.kv.inmemory.descriptors.InMemoryTableDescriptor;
 import org.apache.samza.table.ReadWriteTable;
 import org.apache.samza.table.Table;
-import org.apache.samza.test.harness.IntegrationTestHarness;
+import org.apache.samza.test.framework.TestRunner;
+import org.apache.samza.test.framework.system.descriptors.InMemoryInputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemorySystemDescriptor;
 import org.apache.samza.test.util.ArraySystemFactory;
-import org.apache.samza.test.util.Base64Serializer;
-
-import org.junit.Assert;
 import org.junit.Test;
 
 import static org.apache.samza.test.table.TestTableData.EnrichedPageView;
@@ -62,50 +59,42 @@
 import static org.apache.samza.test.table.TestTableData.ProfileJsonSerde;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertNotNull;
 
 
 /**
  * This test class tests sendTo() and join() for local tables
  */
-public class TestLocalTableEndToEnd extends IntegrationTestHarness {
+public class TestLocalTableEndToEnd {
+  private static final String SYSTEM_NAME = "test";
+  private static final String PAGEVIEW_STREAM = "pageview";
+  private static final String PROFILE_STREAM = "profile";
 
   @Test
-  public void testSendTo() throws Exception {
-
-    int count = 10;
-    Profile[] profiles = TestTableData.generateProfiles(count);
-
-    int partitionCount = 4;
-    Map<String, String> configs = getBaseJobConfig(bootstrapUrl(), zkConnect());
-
-    configs.put("streams.Profile.samza.system", "test");
-    configs.put("streams.Profile.source", Base64Serializer.serialize(profiles));
-    configs.put("streams.Profile.partitionCount", String.valueOf(partitionCount));
-
+  public void testSendTo() {
     MyMapFunction mapFn = new MyMapFunction();
+    StreamApplication app = appDesc -> {
+      Table<KV<Integer, Profile>> table =
+          appDesc.getTable(new InMemoryTableDescriptor<>("t1", KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
+      DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor(SYSTEM_NAME);
+      GenericInputDescriptor<Profile> isd = ksd.getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>());
 
-    final StreamApplication app = appDesc -> {
-
-      Table<KV<Integer, Profile>> table = appDesc.getTable(new InMemoryTableDescriptor("t1",
-          KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
-      DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor("test");
-      GenericInputDescriptor<Profile> isd = ksd.getInputDescriptor("Profile", new NoOpSerde<>());
-
-      appDesc.getInputStream(isd)
-          .map(mapFn)
-          .sendTo(table);
+      appDesc.getInputStream(isd).map(mapFn).sendTo(table);
     };
 
-    Config config = new MapConfig(configs);
-    final LocalApplicationRunner runner = new LocalApplicationRunner(app, config);
-    executeRun(runner, config);
-    runner.waitForFinish();
+    InMemorySystemDescriptor isd = new InMemorySystemDescriptor(SYSTEM_NAME);
+    InMemoryInputDescriptor<Profile> profileStreamDesc = isd.getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>());
 
-    for (int i = 0; i < partitionCount; i++) {
+    int numProfilesPerPartition = 10;
+    int numInputPartitions = 4;
+    Map<Integer, List<Profile>> inputProfiles =
+        TestTableData.generatePartitionedProfiles(numProfilesPerPartition * numInputPartitions, numInputPartitions);
+    TestRunner.of(app).addInputStream(profileStreamDesc, inputProfiles).run(Duration.ofSeconds(10));
+
+    for (int i = 0; i < numInputPartitions; i++) {
       MyMapFunction mapFnCopy = MyMapFunction.getMapFunctionByTask(String.format("Partition %d", i));
-      assertEquals(count, mapFnCopy.received.size());
-      mapFnCopy.received.forEach(p -> Assert.assertTrue(mapFnCopy.table.get(p.getMemberId()) != null));
+      assertEquals(numProfilesPerPartition, mapFnCopy.received.size());
+      mapFnCopy.received.forEach(p -> assertNotNull(mapFnCopy.table.get(p.getMemberId())));
     }
   }
 
@@ -116,52 +105,49 @@
     @Override
     public void describe(StreamApplicationDescriptor appDesc) {
       Table<KV<Integer, Profile>> table = appDesc.getTable(
-          new InMemoryTableDescriptor("t1", KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
-      DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor("test");
-      GenericInputDescriptor<Profile> profileISD = ksd.getInputDescriptor("Profile", new NoOpSerde<>());
+          new InMemoryTableDescriptor<>("t1", KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
+      DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor(SYSTEM_NAME);
+      GenericInputDescriptor<Profile> profileISD = ksd.getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>());
+      profileISD.shouldBootstrap();
       appDesc.getInputStream(profileISD)
-          .map(m -> new KV(m.getMemberId(), m))
+          .map(m -> new KV<>(m.getMemberId(), m))
           .sendTo(table);
 
-      GenericInputDescriptor<PageView> pageViewISD = ksd.getInputDescriptor("PageView", new NoOpSerde<>());
+      GenericInputDescriptor<PageView> pageViewISD = ksd.getInputDescriptor(PAGEVIEW_STREAM, new NoOpSerde<>());
       appDesc.getInputStream(pageViewISD)
           .map(pv -> {
             received.add(pv);
             return pv;
           })
-          .partitionBy(PageView::getMemberId, v -> v, KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()), "p1")
+          .partitionBy(PageView::getMemberId, v -> v, KVSerde.of(new IntegerSerde(), new PageViewJsonSerde()), "p1")
           .join(table, new PageViewToProfileJoinFunction())
           .sink((m, collector, coordinator) -> joined.add(m));
     }
   }
 
   @Test
-  public void testStreamTableJoin() throws Exception {
-
-    int count = 10;
-    PageView[] pageViews = TestTableData.generatePageViews(count);
-    Profile[] profiles = TestTableData.generateProfiles(count);
-
+  public void testStreamTableJoin() {
+    int totalPageViews = 40;
     int partitionCount = 4;
-    Map<String, String> configs = getBaseJobConfig(bootstrapUrl(), zkConnect());
+    Map<Integer, List<PageView>> inputPageViews =
+        TestTableData.generatePartitionedPageViews(totalPageViews, partitionCount);
+    // 10 is the max member id for page views
+    Map<Integer, List<Profile>> inputProfiles =
+        TestTableData.generatePartitionedProfiles(10, partitionCount);
+    InMemorySystemDescriptor isd = new InMemorySystemDescriptor(SYSTEM_NAME);
+    InMemoryInputDescriptor<PageView> pageViewStreamDesc = isd
+        .getInputDescriptor(PAGEVIEW_STREAM, new NoOpSerde<>());
+    InMemoryInputDescriptor<Profile> profileStreamDesc = isd
+        .getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>());
 
-    configs.put("streams.PageView.samza.system", "test");
-    configs.put("streams.PageView.source", Base64Serializer.serialize(pageViews));
-    configs.put("streams.PageView.partitionCount", String.valueOf(partitionCount));
+    TestRunner.of(new StreamTableJoinApp())
+        .addInputStream(pageViewStreamDesc, inputPageViews)
+        .addInputStream(profileStreamDesc, inputProfiles)
+        .run(Duration.ofSeconds(10));
 
-    configs.put("streams.Profile.samza.system", "test");
-    configs.put("streams.Profile.samza.bootstrap", "true");
-    configs.put("streams.Profile.source", Base64Serializer.serialize(profiles));
-    configs.put("streams.Profile.partitionCount", String.valueOf(partitionCount));
-
-    Config config = new MapConfig(configs);
-    final LocalApplicationRunner runner = new LocalApplicationRunner(new StreamTableJoinApp(), config);
-    executeRun(runner, config);
-    runner.waitForFinish();
-
-    assertEquals(count * partitionCount, StreamTableJoinApp.received.size());
-    assertEquals(count * partitionCount, StreamTableJoinApp.joined.size());
-    assertTrue(StreamTableJoinApp.joined.get(0) instanceof EnrichedPageView);
+    assertEquals(totalPageViews, StreamTableJoinApp.received.size());
+    assertEquals(totalPageViews, StreamTableJoinApp.joined.size());
+    assertNotNull(StreamTableJoinApp.joined.get(0));
   }
 
   static class DualStreamTableJoinApp implements StreamApplication {
@@ -178,29 +164,31 @@
       PageViewToProfileJoinFunction joinFn1 = new PageViewToProfileJoinFunction();
       PageViewToProfileJoinFunction joinFn2 = new PageViewToProfileJoinFunction();
 
-      Table<KV<Integer, Profile>> profileTable = appDesc.getTable(new InMemoryTableDescriptor("t1", profileKVSerde));
+      Table<KV<Integer, Profile>> profileTable = appDesc.getTable(new InMemoryTableDescriptor<>("t1", profileKVSerde));
 
-      DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor("test");
-      GenericInputDescriptor<Profile> profileISD1 = ksd.getInputDescriptor("Profile1", new NoOpSerde<>());
-      GenericInputDescriptor<Profile> profileISD2 = ksd.getInputDescriptor("Profile2", new NoOpSerde<>());
+      DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor(SYSTEM_NAME);
+      GenericInputDescriptor<Profile> profileISD1 = ksd.getInputDescriptor(PROFILE_STREAM + "1", new NoOpSerde<>());
+      profileISD1.shouldBootstrap();
+      GenericInputDescriptor<Profile> profileISD2 = ksd.getInputDescriptor(PROFILE_STREAM + "2", new NoOpSerde<>());
+      profileISD2.shouldBootstrap();
       MessageStream<Profile> profileStream1 = appDesc.getInputStream(profileISD1);
       MessageStream<Profile> profileStream2 = appDesc.getInputStream(profileISD2);
 
       profileStream1
           .map(m -> {
             sentToProfileTable1.add(m);
-            return new KV(m.getMemberId(), m);
+            return new KV<>(m.getMemberId(), m);
           })
           .sendTo(profileTable);
       profileStream2
           .map(m -> {
             sentToProfileTable2.add(m);
-            return new KV(m.getMemberId(), m);
+            return new KV<>(m.getMemberId(), m);
           })
           .sendTo(profileTable);
 
-      GenericInputDescriptor<PageView> pageViewISD1 = ksd.getInputDescriptor("PageView1", new NoOpSerde<PageView>());
-      GenericInputDescriptor<PageView> pageViewISD2 = ksd.getInputDescriptor("PageView2", new NoOpSerde<PageView>());
+      GenericInputDescriptor<PageView> pageViewISD1 = ksd.getInputDescriptor(PAGEVIEW_STREAM + "1", new NoOpSerde<>());
+      GenericInputDescriptor<PageView> pageViewISD2 = ksd.getInputDescriptor(PAGEVIEW_STREAM + "2", new NoOpSerde<>());
       MessageStream<PageView> pageViewStream1 = appDesc.getInputStream(pageViewISD1);
       MessageStream<PageView> pageViewStream2 = appDesc.getInputStream(pageViewISD2);
 
@@ -217,45 +205,40 @@
   }
 
   @Test
-  public void testDualStreamTableJoin() throws Exception {
-
-    int count = 10;
-    PageView[] pageViews = TestTableData.generatePageViews(count);
-    Profile[] profiles = TestTableData.generateProfiles(count);
-
+  public void testDualStreamTableJoin() {
+    int totalPageViews = 40;
     int partitionCount = 4;
-    Map<String, String> configs = getBaseJobConfig(bootstrapUrl(), zkConnect());
+    Map<Integer, List<PageView>> inputPageViews1 =
+        TestTableData.generatePartitionedPageViews(totalPageViews, partitionCount);
+    Map<Integer, List<PageView>> inputPageViews2 =
+        TestTableData.generatePartitionedPageViews(totalPageViews, partitionCount);
+    // 10 is the max member id for page views
+    int numProfiles = 10;
+    Map<Integer, List<Profile>> inputProfiles1 = TestTableData.generatePartitionedProfiles(numProfiles, partitionCount);
+    Map<Integer, List<Profile>> inputProfiles2 = TestTableData.generatePartitionedProfiles(numProfiles, partitionCount);
+    InMemorySystemDescriptor isd = new InMemorySystemDescriptor(SYSTEM_NAME);
+    InMemoryInputDescriptor<PageView> pageViewStreamDesc1 = isd
+        .getInputDescriptor(PAGEVIEW_STREAM + "1", new NoOpSerde<>());
+    InMemoryInputDescriptor<PageView> pageViewStreamDesc2 = isd
+        .getInputDescriptor(PAGEVIEW_STREAM + "2", new NoOpSerde<>());
+    InMemoryInputDescriptor<Profile> profileStreamDesc1 = isd
+        .getInputDescriptor(PROFILE_STREAM + "1", new NoOpSerde<>());
+    InMemoryInputDescriptor<Profile> profileStreamDesc2 = isd
+        .getInputDescriptor(PROFILE_STREAM + "2", new NoOpSerde<>());
 
-    configs.put("streams.Profile1.samza.system", "test");
-    configs.put("streams.Profile1.source", Base64Serializer.serialize(profiles));
-    configs.put("streams.Profile1.samza.bootstrap", "true");
-    configs.put("streams.Profile1.partitionCount", String.valueOf(partitionCount));
+    TestRunner.of(new DualStreamTableJoinApp())
+        .addInputStream(pageViewStreamDesc1, inputPageViews1)
+        .addInputStream(pageViewStreamDesc2, inputPageViews2)
+        .addInputStream(profileStreamDesc1, inputProfiles1)
+        .addInputStream(profileStreamDesc2, inputProfiles2)
+        .run(Duration.ofSeconds(10));
 
-    configs.put("streams.Profile2.samza.system", "test");
-    configs.put("streams.Profile2.source", Base64Serializer.serialize(profiles));
-    configs.put("streams.Profile2.samza.bootstrap", "true");
-    configs.put("streams.Profile2.partitionCount", String.valueOf(partitionCount));
-
-    configs.put("streams.PageView1.samza.system", "test");
-    configs.put("streams.PageView1.source", Base64Serializer.serialize(pageViews));
-    configs.put("streams.PageView1.partitionCount", String.valueOf(partitionCount));
-
-    configs.put("streams.PageView2.samza.system", "test");
-    configs.put("streams.PageView2.source", Base64Serializer.serialize(pageViews));
-    configs.put("streams.PageView2.partitionCount", String.valueOf(partitionCount));
-
-    Config config = new MapConfig(configs);
-    final LocalApplicationRunner runner = new LocalApplicationRunner(new DualStreamTableJoinApp(), config);
-    executeRun(runner, config);
-    runner.waitForFinish();
-
-    assertEquals(count * partitionCount, DualStreamTableJoinApp.sentToProfileTable1.size());
-    assertEquals(count * partitionCount, DualStreamTableJoinApp.sentToProfileTable2.size());
-
-    assertEquals(count * partitionCount, DualStreamTableJoinApp.joinedPageViews1.size());
-    assertEquals(count * partitionCount, DualStreamTableJoinApp.joinedPageViews2.size());
-    assertTrue(DualStreamTableJoinApp.joinedPageViews1.get(0) instanceof EnrichedPageView);
-    assertTrue(DualStreamTableJoinApp.joinedPageViews2.get(0) instanceof EnrichedPageView);
+    assertEquals(numProfiles, DualStreamTableJoinApp.sentToProfileTable1.size());
+    assertEquals(numProfiles, DualStreamTableJoinApp.sentToProfileTable2.size());
+    assertEquals(totalPageViews, DualStreamTableJoinApp.joinedPageViews1.size());
+    assertEquals(totalPageViews, DualStreamTableJoinApp.joinedPageViews2.size());
+    assertNotNull(DualStreamTableJoinApp.joinedPageViews1.get(0));
+    assertNotNull(DualStreamTableJoinApp.joinedPageViews2.get(0));
   }
 
   static Map<String, String> getBaseJobConfig(String bootstrapUrl, String zkConnect) {
@@ -283,8 +266,7 @@
   }
 
   private static class MyMapFunction implements MapFunction<Profile, KV<Integer, Profile>> {
-
-    private static Map<String, MyMapFunction> taskToMapFunctionMap = new HashMap<>();
+    private static final Map<String, MyMapFunction> TASK_TO_MAP_FUNCTION_MAP = new HashMap<>();
 
     private transient List<Profile> received;
     private transient ReadWriteTable table;
@@ -294,17 +276,17 @@
       table = context.getTaskContext().getTable("t1");
       this.received = new ArrayList<>();
 
-      taskToMapFunctionMap.put(context.getTaskContext().getTaskModel().getTaskName().getTaskName(), this);
+      TASK_TO_MAP_FUNCTION_MAP.put(context.getTaskContext().getTaskModel().getTaskName().getTaskName(), this);
     }
 
     @Override
     public KV<Integer, Profile> apply(Profile profile) {
       received.add(profile);
-      return new KV(profile.getMemberId(), profile);
+      return new KV<>(profile.getMemberId(), profile);
     }
 
     public static MyMapFunction getMapFunctionByTask(String taskName) {
-      return taskToMapFunctionMap.get(taskName);
+      return TASK_TO_MAP_FUNCTION_MAP.get(taskName);
     }
   }
 
diff --git a/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableWithSideInputsEndToEnd.java b/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableWithSideInputsEndToEnd.java
index 071f65e..34ac29a 100644
--- a/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableWithSideInputsEndToEnd.java
+++ b/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableWithSideInputsEndToEnd.java
@@ -19,128 +19,190 @@
 
 package org.apache.samza.test.table;
 
-import com.google.common.collect.ImmutableList;
 import java.time.Duration;
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
-import org.apache.samza.SamzaException;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.samza.application.SamzaApplication;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.config.MapConfig;
-import org.apache.samza.config.StreamConfig;
+import org.apache.samza.application.TaskApplication;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
 import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptor;
+import org.apache.samza.context.Context;
 import org.apache.samza.operators.KV;
-import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.NoOpSerde;
 import org.apache.samza.storage.kv.Entry;
 import org.apache.samza.storage.kv.descriptors.RocksDbTableDescriptor;
 import org.apache.samza.storage.kv.inmemory.descriptors.InMemoryTableDescriptor;
-import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
+import org.apache.samza.system.IncomingMessageEnvelope;
+import org.apache.samza.system.OutgoingMessageEnvelope;
+import org.apache.samza.system.SystemStream;
+import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
+import org.apache.samza.table.ReadWriteTable;
 import org.apache.samza.table.Table;
+import org.apache.samza.table.descriptors.TableDescriptor;
+import org.apache.samza.task.InitableTask;
+import org.apache.samza.task.MessageCollector;
+import org.apache.samza.task.StreamTask;
+import org.apache.samza.task.StreamTaskFactory;
+import org.apache.samza.task.TaskCoordinator;
+import org.apache.samza.test.framework.StreamAssert;
 import org.apache.samza.test.framework.TestRunner;
 import org.apache.samza.test.framework.system.descriptors.InMemoryInputDescriptor;
 import org.apache.samza.test.framework.system.descriptors.InMemoryOutputDescriptor;
 import org.apache.samza.test.framework.system.descriptors.InMemorySystemDescriptor;
-import org.apache.samza.test.harness.IntegrationTestHarness;
 import org.junit.Test;
 
 import static org.apache.samza.test.table.TestTableData.EnrichedPageView;
 import static org.apache.samza.test.table.TestTableData.PageView;
 import static org.apache.samza.test.table.TestTableData.Profile;
 import static org.apache.samza.test.table.TestTableData.ProfileJsonSerde;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 
 
-public class TestLocalTableWithSideInputsEndToEnd extends IntegrationTestHarness {
+public class TestLocalTableWithSideInputsEndToEnd {
+  private static final String SYSTEM_NAME = "test";
   private static final String PAGEVIEW_STREAM = "pageview";
   private static final String PROFILE_STREAM = "profile";
+  private static final String PROFILE_TABLE = "profile-table";
   private static final String ENRICHED_PAGEVIEW_STREAM = "enrichedpageview";
+  private static final SystemStream OUTPUT_SYSTEM_STREAM = new SystemStream(SYSTEM_NAME, ENRICHED_PAGEVIEW_STREAM);
 
   @Test
-  public void testJoinWithSideInputsTable() {
+  public void testLowLevelJoinWithSideInputsTable() throws InterruptedException {
+    int partitionCount = 4;
+    IntegerSerde integerSerde = new IntegerSerde();
+    // for low-level, need to pre-partition the input in the same way that the profiles are partitioned
+    Map<Integer, List<PageView>> pageViewsPartitionedByMemberId =
+        TestTableData.generatePartitionedPageViews(20, partitionCount)
+            .values()
+            .stream()
+            .flatMap(List::stream)
+            .collect(Collectors.groupingBy(
+              pageView -> Math.abs(Arrays.hashCode(integerSerde.toBytes(pageView.getMemberId()))) % partitionCount));
     runTest(
-        "test",
+        new LowLevelPageViewProfileJoin(),
+        pageViewsPartitionedByMemberId,
+        TestTableData.generatePartitionedProfiles(10, partitionCount));
+  }
+
+  @Test
+  public void testJoinWithSideInputsTable() throws InterruptedException {
+    runTest(
         new PageViewProfileJoin(),
-        Arrays.asList(TestTableData.generatePageViews(10)),
-        Arrays.asList(TestTableData.generateProfiles(10)));
+        TestTableData.generatePartitionedPageViews(20, 4),
+        TestTableData.generatePartitionedProfiles(10, 4));
   }
 
   @Test
-  public void testJoinWithDurableSideInputTable() {
+  public void testJoinWithDurableSideInputTable() throws InterruptedException {
     runTest(
-        "test",
         new DurablePageViewProfileJoin(),
-        Arrays.asList(TestTableData.generatePageViews(5)),
-        Arrays.asList(TestTableData.generateProfiles(5)));
+        TestTableData.generatePartitionedPageViews(20, 4),
+        TestTableData.generatePartitionedProfiles(10, 4));
   }
 
-  private void runTest(String systemName, StreamApplication app, List<PageView> pageViews,
-      List<Profile> profiles) {
-    Map<String, String> configs = new HashMap<>();
-    configs.put(String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, PAGEVIEW_STREAM), systemName);
-    configs.put(String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, PROFILE_STREAM), systemName);
-    configs.put(String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, ENRICHED_PAGEVIEW_STREAM), systemName);
-
-    InMemorySystemDescriptor isd = new InMemorySystemDescriptor(systemName);
-
+  private <T extends ApplicationDescriptor<?>> void runTest(SamzaApplication<T> app,
+      Map<Integer, List<PageView>> pageViews,
+      Map<Integer, List<Profile>> profiles) throws InterruptedException {
+    InMemorySystemDescriptor isd = new InMemorySystemDescriptor(SYSTEM_NAME);
     InMemoryInputDescriptor<PageView> pageViewStreamDesc = isd
-        .getInputDescriptor(PAGEVIEW_STREAM, new NoOpSerde<PageView>());
-
+        .getInputDescriptor(PAGEVIEW_STREAM, new NoOpSerde<>());
     InMemoryInputDescriptor<Profile> profileStreamDesc = isd
-        .getInputDescriptor(PROFILE_STREAM, new NoOpSerde<Profile>());
-
+        .getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>());
     InMemoryOutputDescriptor<EnrichedPageView> outputStreamDesc = isd
-        .getOutputDescriptor(ENRICHED_PAGEVIEW_STREAM, new NoOpSerde<EnrichedPageView>());
+        .getOutputDescriptor(ENRICHED_PAGEVIEW_STREAM, new NoOpSerde<>());
 
-    TestRunner
-        .of(app)
+    TestRunner.of(app)
         .addInputStream(pageViewStreamDesc, pageViews)
         .addInputStream(profileStreamDesc, profiles)
         .addOutputStream(outputStreamDesc, 1)
-        .addConfig(new MapConfig(configs))
-        .run(Duration.ofMillis(100000));
+        .run(Duration.ofSeconds(10));
 
-    try {
-      Map<Integer, List<EnrichedPageView>> result = TestRunner.consumeStream(outputStreamDesc, Duration.ofMillis(1000));
-      List<EnrichedPageView> results = result.values().stream()
-          .flatMap(List::stream)
-          .collect(Collectors.toList());
+    List<EnrichedPageView> expectedEnrichedPageViews = buildExpectedEnrichedPageViews(pageViews, profiles);
+    StreamAssert.containsInAnyOrder(expectedEnrichedPageViews, outputStreamDesc, Duration.ofSeconds(1));
+  }
 
-      List<EnrichedPageView> expectedEnrichedPageviews = pageViews.stream()
-          .flatMap(pv -> profiles.stream()
-              .filter(profile -> pv.memberId == profile.memberId)
-              .map(profile -> new EnrichedPageView(pv.pageKey, profile.memberId, profile.company)))
-          .collect(Collectors.toList());
+  private static List<EnrichedPageView> buildExpectedEnrichedPageViews(Map<Integer, List<PageView>> pageViews,
+      Map<Integer, List<Profile>> profiles) {
+    ImmutableMap.Builder<Integer, Profile> profilesByMemberIdBuilder = new ImmutableMap.Builder<>();
+    profiles.values()
+        .stream()
+        .flatMap(List::stream)
+        .forEach(profile -> profilesByMemberIdBuilder.put(profile.getMemberId(), profile));
+    Map<Integer, Profile> profilesByMemberId = profilesByMemberIdBuilder.build();
+    ImmutableList.Builder<EnrichedPageView> enrichedPageViewsBuilder = new ImmutableList.Builder<>();
+    pageViews.values()
+        .stream()
+        .flatMap(List::stream)
+        .forEach(pageView -> Optional.ofNullable(profilesByMemberId.get(pageView.getMemberId()))
+            .ifPresent(profile -> enrichedPageViewsBuilder.add(
+                new EnrichedPageView(pageView.getPageKey(), profile.getMemberId(), profile.getCompany()))));
+    return enrichedPageViewsBuilder.build();
+  }
 
-      boolean successfulJoin = results.stream().allMatch(expectedEnrichedPageviews::contains);
-      assertEquals("Mismatch between the expected and actual join count", expectedEnrichedPageviews.size(), results.size());
-      assertTrue("Pageview profile join did not succeed for all inputs", successfulJoin);
-    } catch (SamzaException e) {
-      e.printStackTrace();
+  static class LowLevelPageViewProfileJoin implements TaskApplication {
+    @Override
+    public void describe(TaskApplicationDescriptor appDescriptor) {
+      DelegatingSystemDescriptor sd = new DelegatingSystemDescriptor(SYSTEM_NAME);
+      appDescriptor.withInputStream(sd.getInputDescriptor(PAGEVIEW_STREAM, new NoOpSerde<>()));
+      appDescriptor.withInputStream(sd.getInputDescriptor(PROFILE_STREAM, new NoOpSerde<>()));
+
+      TableDescriptor<Integer, Profile, ?> tableDescriptor = new InMemoryTableDescriptor<>(PROFILE_TABLE,
+          KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())).withSideInputs(ImmutableList.of(PROFILE_STREAM))
+          .withSideInputsProcessor((msg, store) -> {
+            Profile profile = (Profile) msg.getMessage();
+            int key = profile.getMemberId();
+            return ImmutableList.of(new Entry<>(key, profile));
+          });
+      appDescriptor.withTable(tableDescriptor);
+
+      appDescriptor.withOutputStream(sd.getOutputDescriptor(ENRICHED_PAGEVIEW_STREAM, new NoOpSerde<>()));
+
+      appDescriptor.withTaskFactory((StreamTaskFactory) PageViewProfileJoinStreamTask::new);
+    }
+  }
+
+  static class PageViewProfileJoinStreamTask implements InitableTask, StreamTask {
+    private ReadWriteTable<Integer, Profile> profileTable;
+
+    @Override
+    public void init(Context context) {
+      this.profileTable = context.getTaskContext().getTable(PROFILE_TABLE);
+    }
+
+    @Override
+    public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) {
+      PageView pageView = (PageView) envelope.getMessage();
+      Profile profile = this.profileTable.get(pageView.getMemberId());
+      if (profile != null) {
+        EnrichedPageView enrichedPageView =
+            new EnrichedPageView(pageView.getPageKey(), profile.getMemberId(), profile.getCompany());
+        collector.send(new OutgoingMessageEnvelope(OUTPUT_SYSTEM_STREAM, enrichedPageView));
+      }
     }
   }
 
   static class PageViewProfileJoin implements StreamApplication {
-    static final String PROFILE_TABLE = "profile-table";
-
     @Override
     public void describe(StreamApplicationDescriptor appDescriptor) {
       Table<KV<Integer, TestTableData.Profile>> table = appDescriptor.getTable(getTableDescriptor());
-      KafkaSystemDescriptor sd =
-          new KafkaSystemDescriptor("test");
+      DelegatingSystemDescriptor sd = new DelegatingSystemDescriptor(SYSTEM_NAME);
       appDescriptor.getInputStream(sd.getInputDescriptor(PAGEVIEW_STREAM, new NoOpSerde<TestTableData.PageView>()))
-          .partitionBy(TestTableData.PageView::getMemberId, v -> v, KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()), "partition-page-view")
+          .partitionBy(TestTableData.PageView::getMemberId, v -> v,
+              KVSerde.of(new IntegerSerde(), new TestTableData.PageViewJsonSerde()), "partition-page-view")
           .join(table, new PageViewToProfileJoinFunction())
           .sendTo(appDescriptor.getOutputStream(sd.getOutputDescriptor(ENRICHED_PAGEVIEW_STREAM, new NoOpSerde<>())));
     }
 
     protected TableDescriptor<Integer, Profile, ?> getTableDescriptor() {
-      return new InMemoryTableDescriptor(PROFILE_TABLE, KVSerde.of(new IntegerSerde(), new ProfileJsonSerde()))
+      return new InMemoryTableDescriptor<>(PROFILE_TABLE, KVSerde.of(new IntegerSerde(), new ProfileJsonSerde()))
           .withSideInputs(ImmutableList.of(PROFILE_STREAM))
           .withSideInputsProcessor((msg, store) -> {
             Profile profile = (Profile) msg.getMessage();
@@ -153,7 +215,7 @@
   static class DurablePageViewProfileJoin extends PageViewProfileJoin {
     @Override
     protected TableDescriptor<Integer, Profile, ?> getTableDescriptor() {
-      return new RocksDbTableDescriptor(PROFILE_TABLE, KVSerde.of(new IntegerSerde(), new ProfileJsonSerde()))
+      return new RocksDbTableDescriptor<>(PROFILE_TABLE, KVSerde.of(new IntegerSerde(), new ProfileJsonSerde()))
           .withSideInputs(ImmutableList.of(PROFILE_STREAM))
           .withSideInputsProcessor((msg, store) -> {
             TestTableData.Profile profile = (TestTableData.Profile) msg.getMessage();
@@ -162,4 +224,4 @@
           });
     }
   }
-}
\ No newline at end of file
+}
diff --git a/samza-test/src/test/java/org/apache/samza/test/table/TestTableData.java b/samza-test/src/test/java/org/apache/samza/test/table/TestTableData.java
index 76c56b0..39f9b02 100644
--- a/samza-test/src/test/java/org/apache/samza/test/table/TestTableData.java
+++ b/samza-test/src/test/java/org/apache/samza/test/table/TestTableData.java
@@ -20,19 +20,29 @@
 package org.apache.samza.test.table;
 
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Random;
-import org.apache.samza.SamzaException;
-import org.apache.samza.config.Config;
-import org.apache.samza.serializers.Serde;
-import org.apache.samza.serializers.SerdeFactory;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import org.apache.samza.SamzaException;
+import org.apache.samza.config.Config;
+import org.apache.samza.serializers.IntegerSerde;
+import org.apache.samza.serializers.Serde;
+import org.apache.samza.serializers.SerdeFactory;
 
 
 public class TestTableData {
+  private static final IntegerSerde INTEGER_SERDE = new IntegerSerde();
 
   public static class PageView implements Serializable {
     @JsonProperty("pageKey")
@@ -205,6 +215,35 @@
     return pageviews;
   }
 
+  /**
+   * Create page views and spread out page views with the same member id across different partitions.
+   * Member ids are spread out like this to make sure that partitionBy operators properly repartition the messages.
+   * Member ids are assigned randomly from [0, 10).
+   *
+   * Example
+   * generatePartitionedPageViews(20, 4) will return:
+   * 0 -> page views with member ids [0, 5)
+   * 1 -> page views with member ids [6, 10)
+   * 2 -> page views with member ids [0, 5)
+   * 3 -> page views with member ids [6, 10)
+   */
+  public static Map<Integer, List<PageView>> generatePartitionedPageViews(int numPageViews, int partitionCount) {
+    Preconditions.checkArgument(numPageViews % partitionCount == 0, "partitionCount must divide numPageViews evenly");
+    int numPerPartition = numPageViews / partitionCount;
+    Random random = new Random();
+    ImmutableMap.Builder<Integer, List<PageView>> pageViewsBuilder = new ImmutableMap.Builder<>();
+    for (int i = 0; i < partitionCount; i++) {
+      pageViewsBuilder.put(i, new ArrayList<>());
+    }
+    Map<Integer, List<PageView>> pageViews = pageViewsBuilder.build();
+    for (int i = 0; i < numPageViews; i++) {
+      String pagekey = PAGEKEYS[random.nextInt(PAGEKEYS.length - 1)];
+      int memberId = i % 10;
+      pageViews.get(i / numPerPartition).add(new PageView(pagekey, memberId));
+    }
+    return pageViews;
+  }
+
   static public PageView[] generatePageViewsWithDistinctKeys(int count) {
     Random random = new Random();
     PageView[] pageviews = new PageView[count];
@@ -227,4 +266,20 @@
     return profiles;
   }
 
+  /**
+   * Create profiles and partition them based on the bytes representation of the member id. This uses the bytes
+   * representation for partitioning because this needs to use the same partition function as the InMemorySystemProducer
+   * (which is used in the test framework) so that table joins can be tested.
+   * One profile for each member id in [0, numProfiles) is created.
+   */
+  public static Map<Integer, List<Profile>> generatePartitionedProfiles(int numProfiles, int partitionCount) {
+    Random random = new Random();
+    return IntStream.range(0, numProfiles)
+        .mapToObj(i -> {
+          String company = COMPANIES[random.nextInt(COMPANIES.length - 1)];
+          return new Profile(i, company);
+        })
+        .collect(Collectors.groupingBy(
+          profile -> Math.abs(Arrays.hashCode(INTEGER_SERDE.toBytes(profile.getMemberId()))) % partitionCount));
+  }
 }
diff --git a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala
index b85e3c5..fb540f7 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala
@@ -41,12 +41,12 @@
   val registry: ReadableMetricsRegistry) extends MetricsHelper with Logging {
 
   private val metricsConfig = new MetricsConfig(config)
-  val containersFromPreviousAttempts = newCounter("container-from-previous-attempt")
+  val containersFromPreviousAttempts = newGauge("container-from-previous-attempt", 0L)
   val reporters = MetricsReporterLoader.getMetricsReporters(metricsConfig, SamzaAppMasterMetrics.sourceName).asScala
   reporters.values.foreach(_.register(SamzaAppMasterMetrics.sourceName, registry))
 
   def setContainersFromPreviousAttempts(containerCount: Int) {
-    containersFromPreviousAttempts.inc(containerCount)
+    containersFromPreviousAttempts.set(containerCount)
   }
 
   def start() {
diff --git a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala
index 7e4565b..237667d 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala
@@ -19,14 +19,11 @@
 
 package org.apache.samza.job.yarn
 
-import java.lang.Boolean
-
 import com.google.common.annotations.VisibleForTesting
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.yarn.api.ApplicationConstants
 import org.apache.hadoop.yarn.api.records.ApplicationId
 import org.apache.samza.SamzaException
-import org.apache.samza.classloader.DependencyIsolationUtils
 import org.apache.samza.config.{Config, JobConfig, ShellCommandConfig, YarnConfig}
 import org.apache.samza.job.ApplicationStatus.{SuccessfulFinish, UnsuccessfulFinish}
 import org.apache.samza.job.{ApplicationStatus, StreamJob}
@@ -46,7 +43,7 @@
   def submit: YarnJob = {
     try {
       val jobConfig = new JobConfig(config)
-      val cmdExec = YarnJob.buildJobCoordinatorCmd(config, jobConfig)
+      val cmdExec = "./__package/bin/run-jc.sh"
       val environment = YarnJob.buildEnvironment(config, this.yarnConfig, jobConfig)
 
       appId = client.submitApplication(
@@ -184,13 +181,6 @@
         Util.envVarEscape(SamzaObjectMapper.getObjectMapper.writeValueAsString(coordinatorSystemConfig))
     }
     envMapBuilder += ShellCommandConfig.ENV_JAVA_OPTS -> Util.envVarEscape(yarnConfig.getAmOpts)
-    val splitDeploymentEnabled = jobConfig.isSplitDeploymentEnabled
-    envMapBuilder += ShellCommandConfig.ENV_SPLIT_DEPLOYMENT_ENABLED -> Util.envVarEscape(Boolean.toString(splitDeploymentEnabled))
-    if (splitDeploymentEnabled) {
-      //split deployment is enabled, so need to specify where the application lib directory is for app resources
-      envMapBuilder += ShellCommandConfig.ENV_APPLICATION_LIB_DIR ->
-        Util.envVarEscape(String.format("./%s/lib", DependencyIsolationUtils.APPLICATION_DIRECTORY))
-    }
     Option.apply(yarnConfig.getAMJavaHome).foreach {
       amJavaHome => envMapBuilder += ShellCommandConfig.ENV_JAVA_HOME -> amJavaHome
     }
@@ -198,18 +188,4 @@
       Util.envVarEscape(config.get(ShellCommandConfig.ADDITIONAL_CLASSPATH_DIR, ""))
     envMapBuilder.result()
   }
-
-  /**
-    * Build the command for the job coordinator execution.
-    * Passing multiple separate config objects so that they can be reused in other places.
-    */
-  @VisibleForTesting
-  private[yarn] def buildJobCoordinatorCmd(config: Config, jobConfig: JobConfig): String = {
-    var cmdExec = "./__package/bin/run-jc.sh" // default location
-    if (jobConfig.isSplitDeploymentEnabled) {
-      cmdExec = "./%s/bin/run-jc.sh" format DependencyIsolationUtils.FRAMEWORK_INFRASTRUCTURE_DIRECTORY
-      logger.info("Using isolated cluster-based job coordinator path: %s" format cmdExec)
-    }
-    cmdExec
-  }
-}
\ No newline at end of file
+}
diff --git a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnJob.java b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnJob.java
index f068800..4e2c4a7 100644
--- a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnJob.java
+++ b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestYarnJob.java
@@ -21,7 +21,6 @@
 import java.io.IOException;
 import java.util.Map;
 import com.google.common.collect.ImmutableMap;
-import org.apache.samza.classloader.DependencyIsolationUtils;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.MapConfig;
@@ -38,19 +37,6 @@
 
 public class TestYarnJob {
   @Test
-  public void testBuildJobCoordinatorCmd() {
-    // cluster-based job coordinator dependency isolation is not enabled; use script from __package directory
-    Config config = new MapConfig();
-    assertEquals("./__package/bin/run-jc.sh", YarnJob$.MODULE$.buildJobCoordinatorCmd(config, new JobConfig(config)));
-
-    // split deployment is enabled; use script from framework infrastructure directory
-    Config splitDeploymentEnabled =
-        new MapConfig(ImmutableMap.of(JobConfig.JOB_SPLIT_DEPLOYMENT_ENABLED, "true"));
-    assertEquals(String.format("./%s/bin/run-jc.sh", DependencyIsolationUtils.FRAMEWORK_INFRASTRUCTURE_DIRECTORY),
-        YarnJob$.MODULE$.buildJobCoordinatorCmd(splitDeploymentEnabled, new JobConfig(splitDeploymentEnabled)));
-  }
-
-  @Test
   public void testBuildEnvironment() throws IOException {
     String amJvmOptions = "-Xmx1g -Dconfig.key='config value'";
     Config config = new MapConfig(new ImmutableMap.Builder<String, String>()
@@ -58,35 +44,12 @@
         .put(JobConfig.JOB_ID, "jobId")
         .put(JobConfig.JOB_COORDINATOR_SYSTEM, "jobCoordinatorSystem")
         .put(YarnConfig.AM_JVM_OPTIONS, amJvmOptions) // needs escaping
-        .put(JobConfig.JOB_SPLIT_DEPLOYMENT_ENABLED, "false")
         .build());
     String expectedCoordinatorStreamConfigStringValue = Util.envVarEscape(SamzaObjectMapper.getObjectMapper()
         .writeValueAsString(CoordinatorStreamUtil.buildCoordinatorStreamConfig(config)));
     Map<String, String> expected = ImmutableMap.of(
         ShellCommandConfig.ENV_COORDINATOR_SYSTEM_CONFIG, expectedCoordinatorStreamConfigStringValue,
         ShellCommandConfig.ENV_JAVA_OPTS, Util.envVarEscape(amJvmOptions),
-        ShellCommandConfig.ENV_SPLIT_DEPLOYMENT_ENABLED, "false",
-        ShellCommandConfig.ENV_ADDITIONAL_CLASSPATH_DIR, "");
-    assertEquals(expected, JavaConverters.mapAsJavaMapConverter(
-        YarnJob$.MODULE$.buildEnvironment(config, new YarnConfig(config), new JobConfig(config))).asJava());
-  }
-
-  @Test
-  public void testBuildEnvironmentJobCoordinatorDependencyIsolationEnabled() throws IOException {
-    Config config = new MapConfig(new ImmutableMap.Builder<String, String>()
-        .put(JobConfig.JOB_NAME, "jobName")
-        .put(JobConfig.JOB_ID, "jobId")
-        .put(JobConfig.JOB_COORDINATOR_SYSTEM, "jobCoordinatorSystem")
-        .put(YarnConfig.AM_JVM_OPTIONS, "")
-        .put(JobConfig.JOB_SPLIT_DEPLOYMENT_ENABLED, "true")
-        .build());
-    String expectedCoordinatorStreamConfigStringValue = Util.envVarEscape(SamzaObjectMapper.getObjectMapper()
-        .writeValueAsString(CoordinatorStreamUtil.buildCoordinatorStreamConfig(config)));
-    Map<String, String> expected = ImmutableMap.of(
-        ShellCommandConfig.ENV_COORDINATOR_SYSTEM_CONFIG, expectedCoordinatorStreamConfigStringValue,
-        ShellCommandConfig.ENV_JAVA_OPTS, "",
-        ShellCommandConfig.ENV_SPLIT_DEPLOYMENT_ENABLED, "true",
-        ShellCommandConfig.ENV_APPLICATION_LIB_DIR, "./__package/lib",
         ShellCommandConfig.ENV_ADDITIONAL_CLASSPATH_DIR, "");
     assertEquals(expected, JavaConverters.mapAsJavaMapConverter(
         YarnJob$.MODULE$.buildEnvironment(config, new YarnConfig(config), new JobConfig(config))).asJava());
@@ -99,7 +62,6 @@
         .put(JobConfig.JOB_ID, "jobId")
         .put(JobConfig.JOB_COORDINATOR_SYSTEM, "jobCoordinatorSystem")
         .put(YarnConfig.AM_JVM_OPTIONS, "")
-        .put(JobConfig.JOB_SPLIT_DEPLOYMENT_ENABLED, "false")
         .put(YarnConfig.AM_JAVA_HOME, "/some/path/to/java/home")
         .build());
     String expectedCoordinatorStreamConfigStringValue = Util.envVarEscape(SamzaObjectMapper.getObjectMapper()
@@ -107,7 +69,6 @@
     Map<String, String> expected = ImmutableMap.of(
         ShellCommandConfig.ENV_COORDINATOR_SYSTEM_CONFIG, expectedCoordinatorStreamConfigStringValue,
         ShellCommandConfig.ENV_JAVA_OPTS, "",
-        ShellCommandConfig.ENV_SPLIT_DEPLOYMENT_ENABLED, "false",
         ShellCommandConfig.ENV_JAVA_HOME, "/some/path/to/java/home",
         ShellCommandConfig.ENV_ADDITIONAL_CLASSPATH_DIR, "");
     assertEquals(expected, JavaConverters.mapAsJavaMapConverter(
@@ -121,15 +82,12 @@
         .put(JobConfig.JOB_ID, "jobId")
         .put(JobConfig.CONFIG_LOADER_FACTORY, "org.apache.samza.config.loaders.PropertiesConfigLoaderFactory")
         .put(YarnConfig.AM_JVM_OPTIONS, "")
-        .put(JobConfig.JOB_SPLIT_DEPLOYMENT_ENABLED, "true")
         .build());
     String expectedSubmissionConfig = Util.envVarEscape(SamzaObjectMapper.getObjectMapper()
         .writeValueAsString(config));
     Map<String, String> expected = ImmutableMap.of(
         ShellCommandConfig.ENV_SUBMISSION_CONFIG, expectedSubmissionConfig,
         ShellCommandConfig.ENV_JAVA_OPTS, "",
-        ShellCommandConfig.ENV_SPLIT_DEPLOYMENT_ENABLED, "true",
-        ShellCommandConfig.ENV_APPLICATION_LIB_DIR, "./__package/lib",
         ShellCommandConfig.ENV_ADDITIONAL_CLASSPATH_DIR, "");
     assertEquals(expected, JavaConverters.mapAsJavaMapConverter(
         YarnJob$.MODULE$.buildEnvironment(config, new YarnConfig(config), new JobConfig(config))).asJava());
@@ -142,7 +100,6 @@
         .put(JobConfig.JOB_ID, "jobId")
         .put(JobConfig.CONFIG_LOADER_FACTORY, "org.apache.samza.config.loaders.PropertiesConfigLoaderFactory")
         .put(YarnConfig.AM_JVM_OPTIONS, "")
-        .put(JobConfig.JOB_SPLIT_DEPLOYMENT_ENABLED, "true")
         .put(ShellCommandConfig.ADDITIONAL_CLASSPATH_DIR, "./sqlapp/lib/*")
         .build());
     String expectedSubmissionConfig = Util.envVarEscape(SamzaObjectMapper.getObjectMapper()
@@ -150,8 +107,6 @@
     Map<String, String> expected = ImmutableMap.of(
         ShellCommandConfig.ENV_SUBMISSION_CONFIG, expectedSubmissionConfig,
         ShellCommandConfig.ENV_JAVA_OPTS, "",
-        ShellCommandConfig.ENV_SPLIT_DEPLOYMENT_ENABLED, "true",
-        ShellCommandConfig.ENV_APPLICATION_LIB_DIR, "./__package/lib",
         ShellCommandConfig.ENV_ADDITIONAL_CLASSPATH_DIR, "./sqlapp/lib/*");
     assertEquals(expected, JavaConverters.mapAsJavaMapConverter(
         YarnJob$.MODULE$.buildEnvironment(config, new YarnConfig(config), new JobConfig(config))).asJava());