diff --git a/lang/cs/Org.Apache.REEF.Client/Avro/AzureBatch/AvroAzureBatchJobSubmissionParameters.cs b/lang/cs/Org.Apache.REEF.Client/Avro/AzureBatch/AvroAzureBatchJobSubmissionParameters.cs
index d0c1e14..88f8365 100644
--- a/lang/cs/Org.Apache.REEF.Client/Avro/AzureBatch/AvroAzureBatchJobSubmissionParameters.cs
+++ b/lang/cs/Org.Apache.REEF.Client/Avro/AzureBatch/AvroAzureBatchJobSubmissionParameters.cs
@@ -5,9 +5,9 @@
 // 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
@@ -22,16 +22,16 @@
 namespace Org.Apache.REEF.Client.Avro.AzureBatch
 {
     /// <summary>
-    /// Used to serialize and deserialize Avro record 
+    /// Used to serialize and deserialize Avro record
     /// org.apache.reef.reef.bridge.client.avro.AvroAzureBatchJobSubmissionParameters.
-    /// This is a (mostly) auto-generated class. 
+    /// This is a (mostly) auto-generated class.
     /// For instructions on how to regenerate, please view the README.md in the same folder.
     /// </summary>
     [Private]
     [DataContract(Namespace = "org.apache.reef.reef.bridge.client.avro")]
     public sealed class AvroAzureBatchJobSubmissionParameters
     {
-        private const string JsonSchema = @"{""type"":""record"",""name"":""org.apache.reef.reef.bridge.client.avro.AvroAzureBatchJobSubmissionParameters"",""doc"":""Job submission parameters used by the Azure Batch runtime"",""fields"":[{""name"":""sharedJobSubmissionParameters"",""type"":{""type"":""record"",""name"":""org.apache.reef.reef.bridge.client.avro.AvroJobSubmissionParameters"",""doc"":""General cross-language job submission parameters shared by all runtimes"",""fields"":[{""name"":""jobId"",""type"":""string""},{""name"":""jobSubmissionFolder"",""type"":""string""}]}},{""name"":""AzureBatchAccountName"",""type"":""string""},{""name"":""AzureBatchAccountUri"",""type"":""string""},{""name"":""AzureBatchPoolId"",""type"":""string""},{""name"":""AzureStorageAccountName"",""type"":""string""},{""name"":""AzureStorageContainerName"",""type"":""string""},{""name"":""AzureBatchPoolDriverPortsList"",""type"":{""type"": ""array"", ""items"": ""string""}}]}";
+        private const string JsonSchema = @"{""type"":""record"",""name"":""org.apache.reef.reef.bridge.client.avro.AvroAzureBatchJobSubmissionParameters"",""doc"":""Job submission parameters used by the Azure Batch runtime"",""fields"":[{""name"":""sharedJobSubmissionParameters"",""type"":{""type"":""record"",""name"":""org.apache.reef.reef.bridge.client.avro.AvroJobSubmissionParameters"",""doc"":""General cross-language job submission parameters shared by all runtimes"",""fields"":[{""name"":""jobId"",""type"":""string""},{""name"":""jobSubmissionFolder"",""type"":""string""}]}},{""name"":""AzureBatchAccountName"",""type"":""string""},{""name"":""AzureBatchAccountUri"",""type"":""string""},{""name"":""AzureBatchPoolId"",""type"":""string""},{""name"":""AzureStorageAccountName"",""type"":""string""},{""name"":""AzureStorageContainerName"",""type"":""string""},{""name"":""AzureBatchPoolDriverPortsList"",""type"":{""type"": ""array"", ""items"": ""string""}},{""name"":""ContainerRegistryServer"",""type"":""string""},{""name"":""ContainerRegistryUsername"",""type"":""string""},{""name"":""ContainerRegistryPassword"",""type"":""string""},{""name"":""ContainerImageName"",""type"":""string""}]}";
 
         /// <summary>
         /// Gets the schema.
@@ -87,6 +87,30 @@
         public IList<string> AzureBatchPoolDriverPortsList { get; set; }
 
         /// <summary>
+        /// Gets or sets the ContainerRegistryServer field.
+        /// </summary>
+        [DataMember]
+        public string ContainerRegistryServer { get; set; }
+
+        /// <summary>
+        /// Gets or sets the ContainerRegistryUsername field.
+        /// </summary>
+        [DataMember]
+        public string ContainerRegistryUsername { get; set; }
+
+        /// <summary>
+        /// Gets or sets the ContainerRegistryPassword field.
+        /// </summary>
+        [DataMember]
+        public string ContainerRegistryPassword { get; set; }
+
+        /// <summary>
+        /// Gets or sets the ContainerImageName field.
+        /// </summary>
+        [DataMember]
+        public string ContainerImageName { get; set; }
+
+        /// <summary>
         /// Initializes a new instance of the <see cref="AvroAzureBatchJobSubmissionParameters"/> class.
         /// </summary>
         public AvroAzureBatchJobSubmissionParameters()
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/AzureBatchDotNetClient.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/AzureBatchDotNetClient.cs
index d2721a4..8bde80c 100644
--- a/lang/cs/Org.Apache.REEF.Client/AzureBatch/AzureBatchDotNetClient.cs
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/AzureBatchDotNetClient.cs
@@ -18,8 +18,9 @@
 using System;
 using System.Threading.Tasks;
 using Org.Apache.REEF.Client.API;
-using Org.Apache.REEF.Client.AzureBatch.Storage;
+using Org.Apache.REEF.Client.API.Parameters;
 using Org.Apache.REEF.Client.AzureBatch;
+using Org.Apache.REEF.Client.AzureBatch.Storage;
 using Org.Apache.REEF.Client.AzureBatch.Util;
 using Org.Apache.REEF.Client.Common;
 using Org.Apache.REEF.Client.YARN.RestClient.DataModel;
@@ -27,8 +28,6 @@
 using Org.Apache.REEF.Tang.Annotations;
 using Org.Apache.REEF.Tang.Interface;
 using Org.Apache.REEF.Utilities.Logging;
-using Org.Apache.REEF.Client.AzureBatch.Parameters;
-using Org.Apache.REEF.Client.API.Parameters;
 
 namespace Org.Apache.REEF.Client.DotNet.AzureBatch
 {
@@ -62,7 +61,7 @@
             AzureBatchService batchService,
             JobJarMaker jobJarMaker,
             //// Those parameters are used in AzureBatchJobSubmissionResult, but could not be injected there.
-            //// It introduces circular injection issues, as all classes constructor inherited from JobSubmissionResult has reference to IREEFClient. 
+            //// It introduces circular injection issues, as all classes constructor inherited from JobSubmissionResult has reference to IREEFClient.
             //// TODO: [REEF-2020] Refactor IJobSubmissionResult Interface and JobSubmissionResult implementation
             [Parameter(typeof(DriverHTTPConnectionRetryInterval))]int retryInterval,
             [Parameter(typeof(DriverHTTPConnectionAttempts))] int numberOfRetries)
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/AzureBatchRuntimeClientConfiguration.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/AzureBatchRuntimeClientConfiguration.cs
index 195e2ce..2c55b5b 100644
--- a/lang/cs/Org.Apache.REEF.Client/AzureBatch/AzureBatchRuntimeClientConfiguration.cs
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/AzureBatchRuntimeClientConfiguration.cs
@@ -15,6 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
+using System;
+using System.Collections.Generic;
+using System.IO;
 using Org.Apache.REEF.Client.API;
 using Org.Apache.REEF.Client.API.Parameters;
 using Org.Apache.REEF.Client.AzureBatch.Parameters;
@@ -22,9 +25,6 @@
 using Org.Apache.REEF.Tang.Formats;
 using Org.Apache.REEF.Tang.Interface;
 using Org.Apache.REEF.Tang.Util;
-using System;
-using System.Collections.Generic;
-using System.IO;
 
 namespace Org.Apache.REEF.Client.AzureBatch
 {
@@ -49,6 +49,11 @@
 
         public static readonly OptionalParameter<IList<string>> AzureBatchPoolDriverPortsList = new OptionalParameter<IList<string>>();
 
+        public static readonly OptionalParameter<string> ContainerRegistryServer = new OptionalParameter<string>();
+        public static readonly OptionalParameter<string> ContainerRegistryUsername = new OptionalParameter<string>();
+        public static readonly OptionalParameter<string> ContainerRegistryPassword = new OptionalParameter<string>();
+        public static readonly OptionalParameter<string> ContainerImageName = new OptionalParameter<string>();
+
         public static ConfigurationModule ConfigurationModule = new AzureBatchRuntimeClientConfiguration()
             .BindImplementation(GenericType<IREEFClient>.Class, GenericType<AzureBatchDotNetClient>.Class)
             .BindNamedParameter(GenericType<AzureBatchAccountUri>.Class, AzureBatchAccountUri)
@@ -61,6 +66,10 @@
             .BindNamedParameter(GenericType<DriverHTTPConnectionRetryInterval>.Class, DriverHTTPConnectionRetryInterval)
             .BindNamedParameter(GenericType<DriverHTTPConnectionAttempts>.Class, DriverHTTPConnectionAttempts)
             .BindNamedParameter(GenericType<AzureBatchPoolDriverPortsList>.Class, AzureBatchPoolDriverPortsList)
+            .BindNamedParameter(GenericType<ContainerRegistryServer>.Class, ContainerRegistryServer)
+            .BindNamedParameter(GenericType<ContainerRegistryUsername>.Class, ContainerRegistryUsername)
+            .BindNamedParameter(GenericType<ContainerRegistryPassword>.Class, ContainerRegistryPassword)
+            .BindNamedParameter(GenericType<ContainerImageName>.Class, ContainerImageName)
             .Build();
 
         public static IConfiguration FromTextFile(string file)
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerImageName.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerImageName.cs
new file mode 100644
index 0000000..d39590d
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerImageName.cs
@@ -0,0 +1,26 @@
+﻿// 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.
+
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.AzureBatch.Parameters
+{
+    [NamedParameter(Documentation = "Docker Container Image Name", DefaultValue = "")]
+    public sealed class ContainerImageName : Name<string>
+    {
+    }
+}
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerRegistryPassword.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerRegistryPassword.cs
new file mode 100644
index 0000000..e75aa9e
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerRegistryPassword.cs
@@ -0,0 +1,26 @@
+﻿// 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.
+
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.AzureBatch.Parameters
+{
+    [NamedParameter(Documentation = "Container Registry Password for Docker Images", DefaultValue = "")]
+    public sealed class ContainerRegistryPassword : Name<string>
+    {
+    }
+}
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerRegistryServer.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerRegistryServer.cs
new file mode 100644
index 0000000..ff9a002
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerRegistryServer.cs
@@ -0,0 +1,26 @@
+﻿// 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.
+
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.AzureBatch.Parameters
+{
+    [NamedParameter(Documentation = "Container Registry Server for Docker Images", DefaultValue = "")]
+    public sealed class ContainerRegistryServer : Name<string>
+    {
+    }
+}
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerRegistryUsername.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerRegistryUsername.cs
new file mode 100644
index 0000000..6e4582d
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Parameters/ContainerRegistryUsername.cs
@@ -0,0 +1,26 @@
+﻿// 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.
+
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.AzureBatch.Parameters
+{
+    [NamedParameter(Documentation = "Container Registry Username for Docker Images", DefaultValue = "")]
+    public sealed class ContainerRegistryUsername : Name<string>
+    {
+    }
+}
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Service/AzureBatchService.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Service/AzureBatchService.cs
index 075ce00..c6deb7b 100644
--- a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Service/AzureBatchService.cs
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Service/AzureBatchService.cs
@@ -17,6 +17,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Azure.Batch;
 using Microsoft.Azure.Batch.Common;
@@ -28,7 +29,7 @@
 
 namespace Org.Apache.REEF.Client.DotNet.AzureBatch
 {
-    public sealed class AzureBatchService : IDisposable
+    internal sealed class AzureBatchService : IDisposable
     {
         private static readonly Logger LOGGER = Logger.GetLogger(typeof(AzureBatchService));
         private static readonly TimeSpan RetryDeltaBackOff = TimeSpan.FromSeconds(5);
@@ -36,24 +37,36 @@
         private const int MaxRetries = 3;
 
         public BatchSharedKeyCredential Credentials { get; private set; }
-        public string PoolId { get; private set; }
+        public string PoolId { get; }
 
-        private BatchClient Client { get; set; }
+        private BatchClient Client { get; }
+        private ContainerRegistryProvider ContainerRegistryProvider { get; }
+        private IList<string> Ports { get; }
+        private ICommandBuilder CommandBuilder { get; }
+        private bool AreContainersEnabled => ContainerRegistryProvider.IsValid();
+
         private bool disposed;
 
         [Inject]
         public AzureBatchService(
+            ContainerRegistryProvider containerRegistryProvider,
+            ICommandBuilder commandBuilder,
             [Parameter(typeof(AzureBatchAccountUri))] string azureBatchAccountUri,
             [Parameter(typeof(AzureBatchAccountName))] string azureBatchAccountName,
             [Parameter(typeof(AzureBatchAccountKey))] string azureBatchAccountKey,
-            [Parameter(typeof(AzureBatchPoolId))] string azureBatchPoolId)
+            [Parameter(typeof(AzureBatchPoolId))] string azureBatchPoolId,
+            [Parameter(typeof(AzureBatchPoolDriverPortsList))] IList<string> ports)
         {
-            BatchSharedKeyCredential credentials = new BatchSharedKeyCredential(azureBatchAccountUri, azureBatchAccountName, azureBatchAccountKey);
+            BatchSharedKeyCredential credentials =
+                new BatchSharedKeyCredential(azureBatchAccountUri, azureBatchAccountName, azureBatchAccountKey);
 
-            this.Client = BatchClient.Open(credentials);
-            this.Credentials = credentials;
-            this.PoolId = azureBatchPoolId;
-            this.Client.CustomBehaviors.Add(new RetryPolicyProvider(new ExponentialRetry(RetryDeltaBackOff, MaxRetries)));
+            Ports = ports;
+            Client = BatchClient.Open(credentials);
+            Credentials = credentials;
+            PoolId = azureBatchPoolId;
+            ContainerRegistryProvider = containerRegistryProvider;
+            Client.CustomBehaviors.Add(new RetryPolicyProvider(new ExponentialRetry(RetryDeltaBackOff, MaxRetries)));
+            CommandBuilder = commandBuilder;
         }
 
         /// <summary>
@@ -61,13 +74,13 @@
         /// </summary>
         public void Dispose()
         {
-            this.Dispose(true);
+            Dispose(true);
             GC.SuppressFinalize(this);
         }
 
         ~AzureBatchService()
         {
-            this.Dispose(false);
+            Dispose(false);
         }
 
         /// <summary>
@@ -75,26 +88,27 @@
         /// </summary>
         private void Dispose(bool disposing)
         {
-            if (this.disposed)
+            if (disposed)
             {
                 return;
             }
 
             if (disposing)
             {
-                this.Client.Dispose();
+                Client.Dispose();
             }
 
-            this.disposed = true;
+            disposed = true;
         }
 
         #region Job related operations
 
         public void CreateJob(string jobId, Uri resourceFile, string commandLine, string storageContainerSAS)
         {
-            CloudJob unboundJob = this.Client.JobOperations.CreateJob();
+            CloudJob unboundJob = Client.JobOperations.CreateJob();
             unboundJob.Id = jobId;
-            unboundJob.PoolInformation = new PoolInformation() { PoolId = this.PoolId };
+            unboundJob.PoolInformation = new PoolInformation() { PoolId = PoolId };
+            unboundJob.JobPreparationTask = CreateJobPreparationTask();
             unboundJob.JobManagerTask = new JobManagerTask()
             {
                 Id = jobId,
@@ -102,26 +116,70 @@
                 RunExclusive = false,
 
                 ResourceFiles = resourceFile != null
-                    ? new List<ResourceFile>() { new ResourceFile(resourceFile.AbsoluteUri, AzureBatchFileNames.GetTaskJarFileName()) }
+                    ? new List<ResourceFile>()
+                    {
+                        new ResourceFile(resourceFile.AbsoluteUri, AzureBatchFileNames.GetTaskJarFileName())
+                    }
                     : new List<ResourceFile>(),
 
-                EnvironmentSettings = new List<EnvironmentSetting> { new EnvironmentSetting(AzureStorageContainerSasToken, storageContainerSAS) },
+                EnvironmentSettings = new List<EnvironmentSetting>
+                {
+                    new EnvironmentSetting(AzureStorageContainerSasToken, storageContainerSAS)
+                },
 
                 // This setting will signal Batch to generate an access token and pass it
                 // to the Job Manager Task (aka the Driver) as an environment variable.
                 // For more info, see
                 // https://docs.microsoft.com/en-us/dotnet/api/microsoft.azure.batch.cloudtask.authenticationtokensettings
-                AuthenticationTokenSettings = new AuthenticationTokenSettings() { Access = AccessScope.Job }
+                AuthenticationTokenSettings = new AuthenticationTokenSettings() { Access = AccessScope.Job },
+                ContainerSettings = CreateTaskContainerSettings(jobId),
             };
 
+            if (AreContainersEnabled)
+            {
+                unboundJob.JobManagerTask.UserIdentity =
+                    new UserIdentity(autoUserSpecification: new AutoUserSpecification(elevationLevel: ElevationLevel.Admin));
+            }
+
             unboundJob.Commit();
 
             LOGGER.Log(Level.Info, "Submitted job {0}, commandLine {1} ", jobId, commandLine);
         }
 
+        private JobPreparationTask CreateJobPreparationTask()
+        {
+            if (!AreContainersEnabled)
+            {
+                return null;
+            }
+
+            return new JobPreparationTask()
+            {
+                Id = "CaptureHostIpAddress",
+                CommandLine = CommandBuilder.CaptureIpAddressCommandLine()
+            };
+        }
+
+        private TaskContainerSettings CreateTaskContainerSettings(string dockerContainerId)
+        {
+            if (!AreContainersEnabled)
+            {
+                return null;
+            }
+
+            string portMappings = Ports
+                .Aggregate(seed: string.Empty, func: (aggregator, port) => $"{aggregator} -p {port}:{port}");
+
+            return new TaskContainerSettings(
+                imageName: ContainerRegistryProvider.ContainerImageName,
+                containerRunOptions:
+                    $"-d --rm --name {dockerContainerId} --env HOST_IP_ADDR_PATH={CommandBuilder.GetIpAddressFilePath()} {portMappings}",
+                registry: ContainerRegistryProvider.GetContainerRegistry());
+        }
+
         public CloudJob GetJob(string jobId, DetailLevel detailLevel)
         {
-            using (Task<CloudJob> getJobTask = this.GetJobAsync(jobId, detailLevel))
+            using (Task<CloudJob> getJobTask = GetJobAsync(jobId, detailLevel))
             {
                 getJobTask.Wait();
                 return getJobTask.Result;
@@ -130,20 +188,20 @@
 
         public Task<CloudJob> GetJobAsync(string jobId, DetailLevel detailLevel)
         {
-            return this.Client.JobOperations.GetJobAsync(jobId, detailLevel);
+            return Client.JobOperations.GetJobAsync(jobId, detailLevel);
         }
 
         public CloudTask GetJobManagerTaskFromJobId(string jobId)
         {
-            string driverTaskId = this.Client.JobOperations.GetJob(jobId).JobManagerTask.Id;
-            return this.Client.JobOperations.GetTask(jobId, driverTaskId);
+            string driverTaskId = Client.JobOperations.GetJob(jobId).JobManagerTask.Id;
+            return Client.JobOperations.GetTask(jobId, driverTaskId);
         }
 
         public ComputeNode GetComputeNodeFromNodeId(string nodeId)
         {
-            return this.Client.PoolOperations.GetComputeNode(this.PoolId, nodeId);
+            return Client.PoolOperations.GetComputeNode(PoolId, nodeId);
         }
 
-        #endregion
+        #endregion Job related operations
     }
-}
\ No newline at end of file
+}
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Service/ContainerRegistryProvider.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Service/ContainerRegistryProvider.cs
new file mode 100644
index 0000000..6d8e3f5
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Service/ContainerRegistryProvider.cs
@@ -0,0 +1,63 @@
+﻿// 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.
+
+using Microsoft.Azure.Batch;
+using Org.Apache.REEF.Client.AzureBatch.Parameters;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.DotNet.AzureBatch
+{
+    public sealed class ContainerRegistryProvider
+    {
+        public string ContainerRegistryServer { get; }
+        public string ContainerRegistryUsername { get; }
+        public string ContainerRegistryPassword { get; }
+        public string ContainerImageName { get; }
+
+        [Inject]
+        public ContainerRegistryProvider(
+            [Parameter(typeof(ContainerRegistryServer))] string containerRegistryServer,
+            [Parameter(typeof(ContainerRegistryUsername))] string containerRegistryUsername,
+            [Parameter(typeof(ContainerRegistryPassword))] string containerRegistryPassword,
+            [Parameter(typeof(ContainerImageName))] string containerImageName
+      )
+        {
+            ContainerRegistryServer = containerRegistryServer;
+            ContainerRegistryUsername = containerRegistryUsername;
+            ContainerRegistryPassword = containerRegistryPassword;
+            ContainerImageName = containerImageName;
+        }
+
+        public bool IsValid()
+        {
+            return !string.IsNullOrEmpty(ContainerRegistryServer);
+        }
+
+        public ContainerRegistry GetContainerRegistry()
+        {
+            if (!IsValid())
+            {
+                return null;
+            }
+
+            return new ContainerRegistry(
+                userName: ContainerRegistryUsername,
+                registryServer: ContainerRegistryServer,
+                password: ContainerRegistryPassword);
+        }
+    }
+}
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/AbstractCommandBuilder.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/AbstractCommandBuilder.cs
index 4363bd7..83e6762 100644
--- a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/AbstractCommandBuilder.cs
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/AbstractCommandBuilder.cs
@@ -5,9 +5,9 @@
 // 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
@@ -63,6 +63,10 @@
             return string.Format(_osCommandFormat, _commandPrefix + sb.ToString());
         }
 
+        public abstract string CaptureIpAddressCommandLine();
+
+        public abstract string GetIpAddressFilePath();
+
         /// <summary>
         /// Returns the driver classpath string which is compatible with the intricacies of the OS.
         /// </summary>
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/ICommandBuilder.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/ICommandBuilder.cs
index 23e83d4..9a4f88c 100644
--- a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/ICommandBuilder.cs
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/ICommandBuilder.cs
@@ -5,9 +5,9 @@
 // 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
@@ -31,5 +31,17 @@
         /// <param name="driverMemory">The memory in megabytes used by driver.</param>
         /// <returns>The command string.</returns>
         string BuildDriverCommand(int driverMemory);
+
+        /// <summary>
+        /// Returns the path to a file where the ip address is persisted.
+        /// </summary>
+        /// <returns>Path to the file.</returns>
+        string GetIpAddressFilePath();
+
+        /// <summary>
+        /// Returns a command line that saves the ip address of the host to a file.
+        /// </summary>
+        /// <returns>The command line string.</returns>
+        string CaptureIpAddressCommandLine();
     }
-}
\ No newline at end of file
+}
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/JobJarMaker.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/JobJarMaker.cs
index 4b02691..f1cbe42 100644
--- a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/JobJarMaker.cs
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/JobJarMaker.cs
@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+using System.Collections.Generic;
+using System.IO;
 using Org.Apache.REEF.Client.API;
 using Org.Apache.REEF.Client.Avro;
 using Org.Apache.REEF.Client.Avro.AzureBatch;
@@ -23,9 +25,6 @@
 using Org.Apache.REEF.Common.Avro;
 using Org.Apache.REEF.Common.Files;
 using Org.Apache.REEF.Tang.Annotations;
-using System;
-using System.Collections.Generic;
-using System.IO;
 
 namespace Org.Apache.REEF.Client.AzureBatch.Util
 {
@@ -37,7 +36,7 @@
         private readonly REEFFileNames _fileNames;
 
         [Inject]
-        JobJarMaker(
+        private JobJarMaker(
             IResourceArchiveFileGenerator resourceArchiveFileGenerator,
             DriverFolderPreparationHelper driverFolderPreparationHelper,
             REEFFileNames fileNames,
@@ -46,7 +45,11 @@
             [Parameter(typeof(AzureBatchPoolId))] string azureBatchPoolId,
             [Parameter(typeof(AzureStorageAccountName))] string azureStorageAccountName,
             [Parameter(typeof(AzureStorageContainerName))] string azureStorageContainerName,
-            [Parameter(typeof(AzureBatchPoolDriverPortsList))] List<string> azureBatchPoolDriverPortsList)
+            [Parameter(typeof(AzureBatchPoolDriverPortsList))] List<string> azureBatchPoolDriverPortsList,
+            [Parameter(typeof(ContainerRegistryServer))] string containerRegistryServer,
+            [Parameter(typeof(ContainerRegistryUsername))] string containerRegistryUsername,
+            [Parameter(typeof(ContainerRegistryPassword))] string containerRegistryPassword,
+            [Parameter(typeof(ContainerImageName))] string containerImageName)
         {
             _resourceArchiveFileGenerator = resourceArchiveFileGenerator;
             _driverFolderPreparationHelper = driverFolderPreparationHelper;
@@ -59,6 +62,10 @@
                 AzureStorageAccountName = azureStorageAccountName,
                 AzureStorageContainerName = azureStorageContainerName,
                 AzureBatchPoolDriverPortsList = azureBatchPoolDriverPortsList,
+                ContainerRegistryServer = containerRegistryServer,
+                ContainerRegistryUsername = containerRegistryUsername,
+                ContainerRegistryPassword = containerRegistryPassword,
+                ContainerImageName = containerImageName,
             };
         }
 
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/LinuxCommandBuilder.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/LinuxCommandBuilder.cs
index eed2163..9780035 100644
--- a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/LinuxCommandBuilder.cs
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/LinuxCommandBuilder.cs
@@ -25,6 +25,7 @@
     {
         private static readonly string CommandPrefix =
             "unzip " + AzureBatchFileNames.GetTaskJarFileName() + " -d 'reef/'" + ";";
+
         private const string ClassPathSeparator = ":";
         private const string OsCommandFormat = "/bin/sh c \"{0}\"";
 
@@ -40,5 +41,16 @@
         {
             throw new NotImplementedException();
         }
+
+        public override string GetIpAddressFilePath()
+        {
+            return "$AZ_BATCH_JOB_PREP_WORKING_DIR/hostip.txt";
+        }
+
+        public override string CaptureIpAddressCommandLine()
+        {
+            string filePath = GetIpAddressFilePath();
+            return $"/bin/bash -c \"rm -f {filePath}; echo `hostname -i` > {filePath}\"";
+        }
     }
 }
\ No newline at end of file
diff --git a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/WindowsCommandBuilder.cs b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/WindowsCommandBuilder.cs
index b7d6d2f..1699d95 100644
--- a/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/WindowsCommandBuilder.cs
+++ b/lang/cs/Org.Apache.REEF.Client/AzureBatch/Util/WindowsCommandBuilder.cs
@@ -5,9 +5,9 @@
 // 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
@@ -27,6 +27,7 @@
           "[System.IO.Compression.ZipFile]::ExtractToDirectory(\\\"$env:AZ_BATCH_TASK_WORKING_DIR\\" +
               AzureBatchFileNames.GetTaskJarFileName() + "\\\", " +
               "\\\"$env:AZ_BATCH_TASK_WORKING_DIR\\reef\\\");";
+
         private const string ClassPathSeparator = ";";
         private const string OsCommandFormat = "powershell.exe /c \"{0}\";";
 
@@ -48,5 +49,15 @@
 
             return string.Format("'{0};'", string.Join(ClassPathSeparator, classpathList));
         }
+
+        public override string GetIpAddressFilePath()
+        {
+            return "%AZ_BATCH_JOB_PREP_WORKING_DIR%\\hostip.txt";
+        }
+
+        public override string CaptureIpAddressCommandLine()
+        {
+            return $"powershell /c \"Set-Content -Path hostip.txt -Value ((Test-Connection -ComputerName $Env:ComputerName -Count 1).IPV4Address.IPAddressToString) -NoNewline -Force\"";
+        }
     }
 }
diff --git a/lang/cs/Org.Apache.REEF.Examples.HelloREEF/HelloREEF.cs b/lang/cs/Org.Apache.REEF.Examples.HelloREEF/HelloREEF.cs
index 4cb1c27..a995a7f 100644
--- a/lang/cs/Org.Apache.REEF.Examples.HelloREEF/HelloREEF.cs
+++ b/lang/cs/Org.Apache.REEF.Examples.HelloREEF/HelloREEF.cs
@@ -16,7 +16,6 @@
 // under the License.
 
 using System;
-using System.Collections.Generic;
 using System.Globalization;
 using Org.Apache.REEF.Client.API;
 using Org.Apache.REEF.Client.AzureBatch;
@@ -98,10 +97,13 @@
                     return LocalRuntimeClientConfiguration.ConfigurationModule
                         .Set(LocalRuntimeClientConfiguration.NumberOfEvaluators, "2")
                         .Build();
+
                 case YARN:
                     return YARNClientConfiguration.ConfigurationModule.Build();
+
                 case YARNRest:
                     return YARNClientConfiguration.ConfigurationModuleYARNRest.Build();
+
                 case HDInsight:
                     // To run against HDInsight please replace placeholders below, with actual values for
                     // blob storage account name and key, container name (available at Azure portal) and HDInsight
@@ -117,6 +119,7 @@
                         .Set(AzureBlobFileSystemConfiguration.AccountName, blobStorageAccountName)
                         .Set(AzureBlobFileSystemConfiguration.AccountKey, blobStorageAccountKey)
                         .Build();
+
                 case AzureBatch:
                     return AzureBatchRuntimeClientConfiguration.ConfigurationModule
                         .Set(AzureBatchRuntimeClientConfiguration.AzureBatchAccountKey, @"##########################################")
@@ -128,9 +131,20 @@
                         .Set(AzureBatchRuntimeClientConfiguration.AzureStorageContainerName, @"###########")
                         //// Extend default retry interval in Azure Batch
                         .Set(AzureBatchRuntimeClientConfiguration.DriverHTTPConnectionRetryInterval, "20000")
-                        //// To allow Driver - Client communication, please specify the ports to use to set up driver http server.
-                        //// These ports must be defined in Azure Batch InBoundNATPool.
-                        .Set(AzureBatchRuntimeClientConfiguration.AzureBatchPoolDriverPortsList, new List<string>(new string[] { "123", "456" }))
+                        //// Following list of ports is required to enable the following options:
+                        //// 1. To enable communication between driver and client:
+                        ////        The ports will be used to set up driver http server endpoint.
+                        ////        In addition, these ports must also be defined in Azure Batch InBoundNATPool to enable communication.
+                        //// 2. To enable communication between docker containers:
+                        ////        These ports will be mapped between the container and host to
+                        ////        allow the communication between the containers.
+                        ////        In addition, the Azure Batch pool must be created as a container pool to use this feature.
+                        // .Set(AzureBatchRuntimeClientConfiguration.AzureBatchPoolDriverPortsList, new List<string> { "2000", "2001", "2002" })
+                        //// Bind to Container Registry properties if present
+                        // .Set(AzureBatchRuntimeClientConfiguration.ContainerRegistryServer, @"###############")
+                        // .Set(AzureBatchRuntimeClientConfiguration.ContainerRegistryUsername, @"###############")
+                        // .Set(AzureBatchRuntimeClientConfiguration.ContainerRegistryPassword, @"###############")
+                        // .Set(AzureBatchRuntimeClientConfiguration.ContainerImageName, @"###############")
                         .Build();
 
                 default:
diff --git a/lang/java/reef-bridge-client/src/main/avro/JobSubmissionParameters.avsc b/lang/java/reef-bridge-client/src/main/avro/JobSubmissionParameters.avsc
index 92f215f..658e4d8 100644
--- a/lang/java/reef-bridge-client/src/main/avro/JobSubmissionParameters.avsc
+++ b/lang/java/reef-bridge-client/src/main/avro/JobSubmissionParameters.avsc
@@ -78,7 +78,11 @@
         { "name": "AzureBatchPoolId", "type": "string" },
         { "name": "AzureStorageAccountName", "type": "string" },
         { "name": "AzureStorageContainerName", "type": "string" },
-        { "name": "AzureBatchPoolDriverPortsList", "type": {"type": "array", "items": "string"}}
+        { "name": "AzureBatchPoolDriverPortsList", "type": {"type": "array", "items": "string"}},
+        { "name": "ContainerRegistryServer", "type": "string" },
+        { "name": "ContainerRegistryUsername", "type": "string" },
+        { "name": "ContainerRegistryPassword", "type": "string" },
+        { "name": "ContainerImageName", "type": "string" }
       ]
   }
 ]
diff --git a/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/AzureBatchBootstrapREEFLauncher.java b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/AzureBatchBootstrapREEFLauncher.java
index 69e39f3..35a2e86 100644
--- a/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/AzureBatchBootstrapREEFLauncher.java
+++ b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/AzureBatchBootstrapREEFLauncher.java
@@ -26,11 +26,7 @@
 import org.apache.reef.runtime.azbatch.AzureBatchClasspathProvider;
 import org.apache.reef.runtime.azbatch.AzureBatchJVMPathProvider;
 import org.apache.reef.runtime.azbatch.client.AzureBatchDriverConfigurationProviderImpl;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchAccountName;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchAccountUri;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchPoolId;
-import org.apache.reef.runtime.azbatch.parameters.AzureStorageAccountName;
-import org.apache.reef.runtime.azbatch.parameters.AzureStorageContainerName;
+import org.apache.reef.runtime.azbatch.parameters.*;
 import org.apache.reef.runtime.azbatch.util.command.CommandBuilder;
 import org.apache.reef.runtime.azbatch.util.command.WindowsCommandBuilder;
 import org.apache.reef.runtime.common.REEFEnvironment;
@@ -43,17 +39,13 @@
 import org.apache.reef.tang.*;
 import org.apache.reef.tang.exceptions.InjectionException;
 import org.apache.reef.wake.remote.RemoteConfiguration;
-import org.apache.reef.wake.remote.ports.ListTcpPortProvider;
-import org.apache.reef.wake.remote.ports.TcpPortProvider;
-import org.apache.reef.wake.remote.ports.parameters.TcpPortList;
+import org.apache.reef.wake.remote.address.LocalAddressProvider;
+import org.apache.reef.wake.remote.ports.parameters.TcpPortSet;
 import org.apache.reef.wake.time.Clock;
 
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -95,20 +87,24 @@
         TANG.newInjector(generateConfiguration(jobSubmissionParameters))
             .getInstance(AzureBatchBootstrapDriverConfigGenerator.class);
 
+    final LocalAddressProvider defaultLocalAddressProvider =
+        Tang.Factory.getTang().newInjector().getInstance(LocalAddressProvider.class);
+
     final JavaConfigurationBuilder launcherConfigBuilder =
         TANG.newConfigurationBuilder()
             .bindNamedParameter(RemoteConfiguration.ManagerName.class, "AzureBatchBootstrapREEFLauncher")
             .bindNamedParameter(RemoteConfiguration.ErrorHandler.class, REEFErrorHandler.class)
             .bindNamedParameter(RemoteConfiguration.MessageCodec.class, REEFMessageCodec.class)
+            .bindNamedParameter(RemoteConfiguration.HostAddress.class, defaultLocalAddressProvider.getLocalAddress())
             .bindSetEntry(Clock.RuntimeStartHandler.class, PIDStoreStartHandler.class);
 
+
     // Check if user has set up preferred ports to use.
     // If set, we prefer will launch driver that binds those ports.
-    final List<String> preferredPorts = asStringList(jobSubmissionParameters.getAzureBatchPoolDriverPortsList());
-
-    if (preferredPorts.size() > 0) {
-      launcherConfigBuilder.bindList(TcpPortList.class, preferredPorts)
-          .bindImplementation(TcpPortProvider.class, ListTcpPortProvider.class);
+    if (jobSubmissionParameters.getAzureBatchPoolDriverPortsList().size() > 0) {
+      for (CharSequence port : jobSubmissionParameters.getAzureBatchPoolDriverPortsList()) {
+        launcherConfigBuilder.bindSetEntry(TcpPortSet.class, port.toString());
+      }
     }
 
     final Configuration launcherConfig = launcherConfigBuilder.build();
@@ -156,17 +152,17 @@
             avroAzureBatchJobSubmissionParameters.getAzureStorageAccountName().toString())
         .bindNamedParameter(AzureStorageContainerName.class,
             avroAzureBatchJobSubmissionParameters.getAzureStorageContainerName().toString())
+        .bindNamedParameter(ContainerRegistryServer.class,
+            avroAzureBatchJobSubmissionParameters.getContainerRegistryServer().toString())
+        .bindNamedParameter(ContainerRegistryUsername.class,
+            avroAzureBatchJobSubmissionParameters.getContainerRegistryUsername().toString())
+        .bindNamedParameter(ContainerRegistryPassword.class,
+            avroAzureBatchJobSubmissionParameters.getContainerRegistryPassword().toString())
+        .bindNamedParameter(ContainerImageName.class,
+            avroAzureBatchJobSubmissionParameters.getContainerImageName().toString())
         .build();
   }
 
-  private static List<String> asStringList(final Collection<? extends CharSequence> list) {
-    final List<String> result = new ArrayList<>(list.size());
-    for (final CharSequence sequence : list) {
-      result.add(sequence.toString());
-    }
-    return result;
-  }
-
   private static RuntimeException fatal(final String msg, final Throwable t) {
     LOG.log(Level.SEVERE, msg, t);
     return new RuntimeException(msg, t);
diff --git a/lang/java/reef-io/src/main/java/org/apache/reef/io/network/naming/NameServerImpl.java b/lang/java/reef-io/src/main/java/org/apache/reef/io/network/naming/NameServerImpl.java
index d52644c..68b75f1 100644
--- a/lang/java/reef-io/src/main/java/org/apache/reef/io/network/naming/NameServerImpl.java
+++ b/lang/java/reef-io/src/main/java/org/apache/reef/io/network/naming/NameServerImpl.java
@@ -33,6 +33,7 @@
 import org.apache.reef.wake.remote.RemoteConfiguration;
 import org.apache.reef.wake.remote.address.LocalAddressProvider;
 import org.apache.reef.wake.remote.impl.TransportEvent;
+import org.apache.reef.wake.remote.ports.TcpPortProvider;
 import org.apache.reef.wake.remote.transport.Transport;
 import org.apache.reef.wake.remote.transport.netty.NettyMessagingTransport;
 import org.apache.reef.webserver.ReefEventStateManager;
@@ -48,6 +49,7 @@
  */
 public final class NameServerImpl implements NameServer {
 
+  private static final String UNKNOWN_HOST_NAME = "##UNKNOWN##";
   private static final Logger LOG = Logger.getLogger(NameServer.class.getName());
 
   private final Transport transport;
@@ -64,8 +66,10 @@
    */
   @Inject
   private NameServerImpl(
+      @Parameter(RemoteConfiguration.HostAddress.class) final String hostAddress,
       @Parameter(NameServerParameters.NameServerPort.class) final int port,
       @Parameter(NameServerParameters.NameServerIdentifierFactory.class) final IdentifierFactory factory,
+      final TcpPortProvider portProvider,
       final LocalAddressProvider localAddressProvider) {
 
     final Injector injector = Tang.Factory.getTang().newInjector();
@@ -75,8 +79,10 @@
     final Codec<NamingMessage> codec = NamingCodecFactory.createFullCodec(factory);
     final EventHandler<NamingMessage> handler = createEventHandler(codec);
 
-    injector.bindVolatileParameter(RemoteConfiguration.HostAddress.class, localAddressProvider.getLocalAddress());
+    String host = UNKNOWN_HOST_NAME.equals(hostAddress) ? localAddressProvider.getLocalAddress() : hostAddress;
+    injector.bindVolatileParameter(RemoteConfiguration.HostAddress.class, host);
     injector.bindVolatileParameter(RemoteConfiguration.Port.class, port);
+    injector.bindVolatileInstance(TcpPortProvider.class, portProvider);
     injector.bindVolatileParameter(RemoteConfiguration.RemoteServerStage.class,
         new SyncStage<>(new NamingServerHandler(handler, codec)));
 
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchDriverConfigurationProviderImpl.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchDriverConfigurationProviderImpl.java
index 0887e9d..adab2cb 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchDriverConfigurationProviderImpl.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchDriverConfigurationProviderImpl.java
@@ -21,20 +21,25 @@
 import org.apache.reef.annotations.audience.Private;
 import org.apache.reef.runtime.azbatch.driver.AzureBatchDriverConfiguration;
 import org.apache.reef.runtime.azbatch.driver.RuntimeIdentifier;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchAccountName;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchPoolId;
+import org.apache.reef.runtime.azbatch.parameters.*;
+import org.apache.reef.runtime.azbatch.util.batch.AzureBatchHelper;
+import org.apache.reef.runtime.azbatch.util.batch.ContainerRegistryProvider;
 import org.apache.reef.runtime.azbatch.util.command.CommandBuilder;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchAccountUri;
-import org.apache.reef.runtime.azbatch.parameters.AzureStorageAccountName;
-import org.apache.reef.runtime.azbatch.parameters.AzureStorageContainerName;
 import org.apache.reef.runtime.common.client.DriverConfigurationProvider;
 import org.apache.reef.runtime.common.parameters.JVMHeapSlack;
 import org.apache.reef.tang.Configuration;
 import org.apache.reef.tang.Configurations;
 import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.tang.formats.ConfigurationModuleBuilder;
+import org.apache.reef.wake.remote.address.ContainerBasedLocalAddressProvider;
+import org.apache.reef.wake.remote.address.LocalAddressProvider;
+import org.apache.reef.wake.remote.ports.SetTcpPortProvider;
+import org.apache.reef.wake.remote.ports.TcpPortProvider;
+import org.apache.reef.wake.remote.ports.parameters.TcpPortSet;
 
 import javax.inject.Inject;
 import java.net.URI;
+import java.util.Set;
 
 /**
  * Configuration provider for the Azure Batch runtime.
@@ -48,7 +53,9 @@
   private final String azureBatchPoolId;
   private final String azureStorageAccountName;
   private final String azureStorageContainerName;
+  private final ContainerRegistryProvider containerRegistryProvider;
   private final CommandBuilder commandBuilder;
+  private final Set<String> tcpPortSet;
 
   @Inject
   private AzureBatchDriverConfigurationProviderImpl(
@@ -58,6 +65,8 @@
       @Parameter(AzureBatchPoolId.class) final String azureBatchPoolId,
       @Parameter(AzureStorageAccountName.class) final String azureStorageAccountName,
       @Parameter(AzureStorageContainerName.class) final String azureStorageContainerName,
+      @Parameter(TcpPortSet.class) final Set<Integer> tcpPortSet,
+      final ContainerRegistryProvider containerRegistryProvider,
       final CommandBuilder commandBuilder) {
     this.jvmSlack = jvmSlack;
     this.azureBatchAccountUri = azureBatchAccountUri;
@@ -65,7 +74,11 @@
     this.azureBatchPoolId = azureBatchPoolId;
     this.azureStorageAccountName = azureStorageAccountName;
     this.azureStorageContainerName = azureStorageContainerName;
+    this.containerRegistryProvider = containerRegistryProvider;
     this.commandBuilder = commandBuilder;
+
+    // Binding a parameter to a set is only allowed for strings, so we cast to strings.
+    this.tcpPortSet = AzureBatchHelper.toStringSet(tcpPortSet);
   }
 
   /**
@@ -82,19 +95,36 @@
                                               final String clientRemoteId,
                                               final String jobId,
                                               final Configuration applicationConfiguration) {
-    return Configurations.merge(
-        AzureBatchDriverConfiguration.CONF.getBuilder()
-            .bindImplementation(CommandBuilder.class, this.commandBuilder.getClass()).build()
-            .set(AzureBatchDriverConfiguration.JOB_IDENTIFIER, jobId)
-            .set(AzureBatchDriverConfiguration.CLIENT_REMOTE_IDENTIFIER, clientRemoteId)
-            .set(AzureBatchDriverConfiguration.JVM_HEAP_SLACK, this.jvmSlack)
-            .set(AzureBatchDriverConfiguration.RUNTIME_NAME, RuntimeIdentifier.RUNTIME_NAME)
-            .set(AzureBatchDriverConfiguration.AZURE_BATCH_ACCOUNT_URI, this.azureBatchAccountUri)
-            .set(AzureBatchDriverConfiguration.AZURE_BATCH_ACCOUNT_NAME, this.azureBatchAccountName)
-            .set(AzureBatchDriverConfiguration.AZURE_BATCH_POOL_ID, this.azureBatchPoolId)
-            .set(AzureBatchDriverConfiguration.AZURE_STORAGE_ACCOUNT_NAME, this.azureStorageAccountName)
-            .set(AzureBatchDriverConfiguration.AZURE_STORAGE_CONTAINER_NAME, this.azureStorageContainerName)
-            .build(),
-        applicationConfiguration);
+    ConfigurationModuleBuilder driverConfigurationBuilder = AzureBatchDriverConfiguration.CONF.getBuilder()
+        .bindImplementation(CommandBuilder.class, this.commandBuilder.getClass());
+
+    // If using docker containers, then use a different set of bindings
+    if (this.containerRegistryProvider.isValid()) {
+      driverConfigurationBuilder = driverConfigurationBuilder
+          .bindImplementation(LocalAddressProvider.class, ContainerBasedLocalAddressProvider.class)
+          .bindImplementation(TcpPortProvider.class, SetTcpPortProvider.class);
+    }
+
+    final Configuration driverConfiguration = driverConfigurationBuilder.build()
+        .set(AzureBatchDriverConfiguration.JOB_IDENTIFIER, jobId)
+        .set(AzureBatchDriverConfiguration.CLIENT_REMOTE_IDENTIFIER, clientRemoteId)
+        .set(AzureBatchDriverConfiguration.JVM_HEAP_SLACK, this.jvmSlack)
+        .set(AzureBatchDriverConfiguration.RUNTIME_NAME, RuntimeIdentifier.RUNTIME_NAME)
+        .set(AzureBatchDriverConfiguration.AZURE_BATCH_ACCOUNT_URI, this.azureBatchAccountUri)
+        .set(AzureBatchDriverConfiguration.AZURE_BATCH_ACCOUNT_NAME, this.azureBatchAccountName)
+        .set(AzureBatchDriverConfiguration.AZURE_BATCH_POOL_ID, this.azureBatchPoolId)
+        .set(AzureBatchDriverConfiguration.AZURE_STORAGE_ACCOUNT_NAME, this.azureStorageAccountName)
+        .set(AzureBatchDriverConfiguration.AZURE_STORAGE_CONTAINER_NAME, this.azureStorageContainerName)
+        .set(AzureBatchDriverConfiguration.CONTAINER_REGISTRY_SERVER,
+            this.containerRegistryProvider.getContainerRegistryServer())
+        .set(AzureBatchDriverConfiguration.CONTAINER_REGISTRY_USERNAME,
+            this.containerRegistryProvider.getContainerRegistryUsername())
+        .set(AzureBatchDriverConfiguration.CONTAINER_REGISTRY_PASSWORD,
+            this.containerRegistryProvider.getContainerRegistryPassword())
+        .set(AzureBatchDriverConfiguration.CONTAINER_IMAGE_NAME,
+            this.containerRegistryProvider.getContainerImageName())
+        .setMultiple(AzureBatchDriverConfiguration.TCP_PORT_SET, this.tcpPortSet)
+        .build();
+    return Configurations.merge(driverConfiguration, applicationConfiguration);
   }
 }
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchJobSubmissionHandler.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchJobSubmissionHandler.java
index cbb8a8c..95e2d24 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchJobSubmissionHandler.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchJobSubmissionHandler.java
@@ -93,7 +93,7 @@
    */
   @Override
   public void close() throws Exception {
-    LOG.log(Level.INFO, "Closing " + AzureBatchJobSubmissionHandler.class.getName());
+    LOG.log(Level.INFO, "Closing {0}", AzureBatchJobSubmissionHandler.class.getName());
   }
 
   /**
@@ -147,7 +147,7 @@
 
   private String createApplicationId(final JobSubmissionEvent jobSubmissionEvent) {
     String uuid = UUID.randomUUID().toString();
-    String jobIdentifier  = jobSubmissionEvent.getIdentifier();
+    String jobIdentifier = jobSubmissionEvent.getIdentifier();
     String jobNameShort = jobIdentifier.length() + 1 + uuid.length() < MAX_CHARS_JOB_NAME ?
         jobIdentifier : jobIdentifier.substring(0, MAX_CHARS_JOB_NAME - uuid.length() - 1);
     return jobNameShort + "-" + uuid;
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfiguration.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfiguration.java
index 541f8c4..7489211 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfiguration.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfiguration.java
@@ -23,6 +23,7 @@
 import org.apache.reef.tang.formats.AvroConfigurationSerializer;
 
 import org.apache.reef.tang.formats.ConfigurationModuleBuilder;
+import org.apache.reef.tang.formats.OptionalParameter;
 import org.apache.reef.tang.formats.RequiredParameter;
 
 import java.io.File;
@@ -75,6 +76,31 @@
   public static final RequiredParameter<String> AZURE_STORAGE_CONTAINER_NAME = new RequiredParameter<>();
 
   /**
+   * Container Registry server.
+   */
+  public static final OptionalParameter<String> CONTAINER_REGISTRY_SERVER = new OptionalParameter<>();
+
+  /**
+   * Container Registry username.
+   */
+  public static final OptionalParameter<String> CONTAINER_REGISTRY_USERNAME = new OptionalParameter<>();
+
+  /**
+   * Container Registry password.
+   */
+  public static final OptionalParameter<String> CONTAINER_REGISTRY_PASSWORD = new OptionalParameter<>();
+
+  /**
+   * Container Image name.
+   */
+  public static final OptionalParameter<String> CONTAINER_IMAGE_NAME = new OptionalParameter<>();
+
+  /**
+   * Set of tcp ports.
+   */
+  public static final OptionalParameter<Integer> TCP_PORT_SET = new OptionalParameter<>();
+
+  /**
    * Create a {@link Configuration} object from an Avro configuration file.
    *
    * @param file the configuration file.
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfigurationCreator.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfigurationCreator.java
index 0d8860b..aa6e6d3 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfigurationCreator.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfigurationCreator.java
@@ -19,18 +19,13 @@
 package org.apache.reef.runtime.azbatch.client;
 
 import org.apache.reef.annotations.audience.Private;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchAccountKey;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchAccountName;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchAccountUri;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchPoolId;
+import org.apache.reef.runtime.azbatch.parameters.*;
 import org.apache.reef.runtime.azbatch.util.command.CommandBuilder;
 import org.apache.reef.runtime.azbatch.util.command.LinuxCommandBuilder;
 import org.apache.reef.runtime.azbatch.util.command.WindowsCommandBuilder;
-import org.apache.reef.runtime.azbatch.parameters.AzureStorageAccountKey;
-import org.apache.reef.runtime.azbatch.parameters.AzureStorageAccountName;
-import org.apache.reef.runtime.azbatch.parameters.AzureStorageContainerName;
 import org.apache.reef.tang.formats.ConfigurationModule;
 import org.apache.reef.tang.formats.ConfigurationModuleBuilder;
+import org.apache.reef.wake.remote.ports.parameters.TcpPortSet;
 
 /**
  * Class that builds the ConfigurationModule for Azure Batch runtime.
@@ -49,27 +44,35 @@
    * @param isWindows true if Azure Batch pool nodes run Windows, false otherwise.
    * @return the configuration module object.
    */
-  public static ConfigurationModule getOrCreateAzureBatchRuntimeConfiguration(final boolean isWindows) {
+  public static ConfigurationModule getOrCreateAzureBatchRuntimeConfiguration(
+      final boolean isWindows) {
 
     if (AzureBatchRuntimeConfigurationCreator.conf == null) {
       ConfigurationModuleBuilder builder = AzureBatchRuntimeConfigurationStatic.CONF;
-      ConfigurationModule module;
+
       if (isWindows) {
-        module = builder.bindImplementation(CommandBuilder.class, WindowsCommandBuilder.class).build();
+        builder = builder.bindImplementation(CommandBuilder.class, WindowsCommandBuilder.class);
       } else {
-        module = builder.bindImplementation(CommandBuilder.class, LinuxCommandBuilder.class).build();
+        builder = builder.bindImplementation(CommandBuilder.class, LinuxCommandBuilder.class);
       }
 
       AzureBatchRuntimeConfigurationCreator.conf = new AzureBatchRuntimeConfiguration()
-          .merge(module)
+          .merge(builder.build())
           .bindNamedParameter(AzureBatchAccountName.class, AzureBatchRuntimeConfiguration.AZURE_BATCH_ACCOUNT_NAME)
           .bindNamedParameter(AzureBatchAccountUri.class, AzureBatchRuntimeConfiguration.AZURE_BATCH_ACCOUNT_URI)
           .bindNamedParameter(AzureBatchAccountKey.class, AzureBatchRuntimeConfiguration.AZURE_BATCH_ACCOUNT_KEY)
           .bindNamedParameter(AzureBatchPoolId.class, AzureBatchRuntimeConfiguration.AZURE_BATCH_POOL_ID)
           .bindNamedParameter(AzureStorageAccountName.class, AzureBatchRuntimeConfiguration.AZURE_STORAGE_ACCOUNT_NAME)
           .bindNamedParameter(AzureStorageAccountKey.class, AzureBatchRuntimeConfiguration.AZURE_STORAGE_ACCOUNT_KEY)
+          .bindNamedParameter(ContainerRegistryServer.class, AzureBatchRuntimeConfiguration.CONTAINER_REGISTRY_SERVER)
+          .bindNamedParameter(
+              ContainerRegistryUsername.class, AzureBatchRuntimeConfiguration.CONTAINER_REGISTRY_USERNAME)
+          .bindNamedParameter(
+              ContainerRegistryPassword.class, AzureBatchRuntimeConfiguration.CONTAINER_REGISTRY_PASSWORD)
+          .bindNamedParameter(ContainerImageName.class, AzureBatchRuntimeConfiguration.CONTAINER_IMAGE_NAME)
           .bindNamedParameter(
               AzureStorageContainerName.class, AzureBatchRuntimeConfiguration.AZURE_STORAGE_CONTAINER_NAME)
+          .bindSetEntry(TcpPortSet.class, AzureBatchRuntimeConfiguration.TCP_PORT_SET)
           .build();
     }
 
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfigurationProvider.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfigurationProvider.java
index c78e9e3..ce62be4 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfigurationProvider.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/client/AzureBatchRuntimeConfigurationProvider.java
@@ -20,17 +20,20 @@
 
 import org.apache.reef.annotations.audience.Public;
 import org.apache.reef.runtime.azbatch.parameters.*;
+import org.apache.reef.runtime.azbatch.util.batch.AzureBatchHelper;
+import org.apache.reef.runtime.azbatch.util.batch.ContainerRegistryProvider;
 import org.apache.reef.tang.Configuration;
 import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.wake.remote.ports.parameters.TcpPortSet;
 
 import javax.inject.Inject;
+import java.util.Set;
 
 /**
  * Class that provides the runtime configuration for Azure Batch.
  */
 @Public
 public final class AzureBatchRuntimeConfigurationProvider {
-
   private final String azureBatchAccountName;
   private final String azureBatchAccountKey;
   private final String azureBatchAccountUri;
@@ -38,7 +41,9 @@
   private final String azureStorageAccountName;
   private final String azureStorageAccountKey;
   private final String azureStorageContainerName;
+  private final ContainerRegistryProvider containerRegistryProvider;
   private final Boolean isWindows;
+  private final Set<String> tcpPortSet;
 
   /**
    * Private constructor.
@@ -52,7 +57,9 @@
       @Parameter(AzureStorageAccountName.class) final String azureStorageAccountName,
       @Parameter(AzureStorageAccountKey.class) final String azureStorageAccountKey,
       @Parameter(AzureStorageContainerName.class) final String azureStorageContainerName,
-      @Parameter(IsWindows.class) final Boolean isWindows) {
+      @Parameter(IsWindows.class) final Boolean isWindows,
+      @Parameter(TcpPortSet.class) final Set<Integer> tcpPortSet,
+      final ContainerRegistryProvider containerRegistryProvider) {
     this.azureBatchAccountName = azureBatchAccountName;
     this.azureBatchAccountKey = azureBatchAccountKey;
     this.azureBatchAccountUri = azureBatchAccountUri;
@@ -60,7 +67,11 @@
     this.azureStorageAccountName = azureStorageAccountName;
     this.azureStorageAccountKey = azureStorageAccountKey;
     this.azureStorageContainerName = azureStorageContainerName;
+    this.containerRegistryProvider = containerRegistryProvider;
     this.isWindows = isWindows;
+
+    // Binding a parameter to a set is only allowed for strings, so we cast to strings.
+    this.tcpPortSet = AzureBatchHelper.toStringSet(tcpPortSet);
   }
 
   public Configuration getAzureBatchRuntimeConfiguration() {
@@ -73,6 +84,15 @@
         .set(AzureBatchRuntimeConfiguration.AZURE_STORAGE_ACCOUNT_NAME, this.azureStorageAccountName)
         .set(AzureBatchRuntimeConfiguration.AZURE_STORAGE_ACCOUNT_KEY, this.azureStorageAccountKey)
         .set(AzureBatchRuntimeConfiguration.AZURE_STORAGE_CONTAINER_NAME, this.azureStorageContainerName)
+        .set(AzureBatchRuntimeConfiguration.CONTAINER_REGISTRY_SERVER,
+            this.containerRegistryProvider.getContainerRegistryServer())
+        .set(AzureBatchRuntimeConfiguration.CONTAINER_REGISTRY_USERNAME,
+            this.containerRegistryProvider.getContainerRegistryUsername())
+        .set(AzureBatchRuntimeConfiguration.CONTAINER_REGISTRY_PASSWORD,
+            this.containerRegistryProvider.getContainerRegistryPassword())
+        .set(AzureBatchRuntimeConfiguration.CONTAINER_IMAGE_NAME,
+            this.containerRegistryProvider.getContainerImageName())
+        .setMultiple(AzureBatchRuntimeConfiguration.TCP_PORT_SET, this.tcpPortSet)
         .build();
   }
 }
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/driver/AzureBatchDriverConfiguration.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/driver/AzureBatchDriverConfiguration.java
index 9a096be..b8b8f18 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/driver/AzureBatchDriverConfiguration.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/driver/AzureBatchDriverConfiguration.java
@@ -41,6 +41,7 @@
 import org.apache.reef.tang.formats.ConfigurationModuleBuilder;
 import org.apache.reef.tang.formats.OptionalParameter;
 import org.apache.reef.tang.formats.RequiredParameter;
+import org.apache.reef.wake.remote.ports.parameters.TcpPortSet;
 
 /**
  * ConfigurationModule to create Azure Batch Driver configurations.
@@ -95,6 +96,31 @@
   public static final RequiredParameter<String> AZURE_STORAGE_CONTAINER_NAME = new RequiredParameter<>();
 
   /**
+   * Container Registry Server.
+   */
+  public static final OptionalParameter<String> CONTAINER_REGISTRY_SERVER = new OptionalParameter<>();
+
+  /**
+   * Container Registry Username.
+   */
+  public static final OptionalParameter<String> CONTAINER_REGISTRY_USERNAME = new OptionalParameter<>();
+
+  /**
+   * Container Registry password.
+   */
+  public static final OptionalParameter<String> CONTAINER_REGISTRY_PASSWORD = new OptionalParameter<>();
+
+  /**
+   * Container Image name.
+   */
+  public static final OptionalParameter<String> CONTAINER_IMAGE_NAME = new OptionalParameter<>();
+
+  /**
+   * Set of TCP Ports.
+   */
+  public static final OptionalParameter<Integer> TCP_PORT_SET = new OptionalParameter<>();
+
+  /**
    * The fraction of the container memory NOT to use for the Java Heap.
    */
   public static final OptionalParameter<Double> JVM_HEAP_SLACK = new OptionalParameter<>();
@@ -115,6 +141,12 @@
       .bindNamedParameter(AzureStorageAccountName.class, AZURE_STORAGE_ACCOUNT_NAME)
       .bindNamedParameter(AzureStorageContainerName.class, AZURE_STORAGE_CONTAINER_NAME)
 
+      // Bind Azure Container Parameters
+      .bindNamedParameter(ContainerRegistryServer.class, CONTAINER_REGISTRY_SERVER)
+      .bindNamedParameter(ContainerRegistryUsername.class, CONTAINER_REGISTRY_USERNAME)
+      .bindNamedParameter(ContainerRegistryPassword.class, CONTAINER_REGISTRY_PASSWORD)
+      .bindNamedParameter(ContainerImageName.class, CONTAINER_IMAGE_NAME)
+
       // Bind the fields bound in AbstractDriverRuntimeConfiguration
       .bindNamedParameter(JobIdentifier.class, JOB_IDENTIFIER)
       .bindNamedParameter(LaunchID.class, JOB_IDENTIFIER)
@@ -125,5 +157,6 @@
       .bindImplementation(RuntimeClasspathProvider.class, AzureBatchClasspathProvider.class)
       .bindImplementation(RuntimePathProvider.class, AzureBatchJVMPathProvider.class)
       .bindSetEntry(DefinedRuntimes.class, RUNTIME_NAME)
+      .bindSetEntry(TcpPortSet.class, TCP_PORT_SET)
       .build();
 }
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/driver/AzureBatchEvaluatorShimConfigurationProvider.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/driver/AzureBatchEvaluatorShimConfigurationProvider.java
index 85b9103..e3d8b1b 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/driver/AzureBatchEvaluatorShimConfigurationProvider.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/driver/AzureBatchEvaluatorShimConfigurationProvider.java
@@ -20,10 +20,15 @@
 
 import org.apache.reef.annotations.audience.Private;
 import org.apache.reef.runtime.azbatch.evaluator.EvaluatorShimConfiguration;
+import org.apache.reef.runtime.azbatch.util.batch.AzureBatchHelper;
+import org.apache.reef.runtime.azbatch.util.batch.ContainerRegistryProvider;
 import org.apache.reef.runtime.common.utils.RemoteManager;
 import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.wake.remote.ports.parameters.TcpPortSet;
 
 import javax.inject.Inject;
+import java.util.Set;
 
 /**
  * Configuration provider for the Azure Batch evaluator shim.
@@ -32,10 +37,18 @@
 public class AzureBatchEvaluatorShimConfigurationProvider {
 
   private final RemoteManager remoteManager;
+  private final Set<String> tcpPortSet;
+  private final ContainerRegistryProvider containerRegistryProvider;
 
   @Inject
-  AzureBatchEvaluatorShimConfigurationProvider(final RemoteManager remoteManager) {
+  AzureBatchEvaluatorShimConfigurationProvider(
+      @Parameter(TcpPortSet.class) final Set<Integer> tcpPortSet,
+      final ContainerRegistryProvider containerRegistryProvider,
+      final RemoteManager remoteManager) {
     this.remoteManager = remoteManager;
+    this.containerRegistryProvider = containerRegistryProvider;
+    // Binding a parameter to a set is only allowed for strings, so we cast to strings.
+    this.tcpPortSet = AzureBatchHelper.toStringSet(tcpPortSet);
   }
 
   /**
@@ -46,9 +59,13 @@
    * @return A {@link Configuration} object needed to launch the evaluator shim.
    */
   public Configuration getConfiguration(final String containerId) {
-    return EvaluatorShimConfiguration.CONF
+
+    return EvaluatorShimConfiguration
+        .getConfigurationModule(this.containerRegistryProvider.isValid())
         .set(EvaluatorShimConfiguration.DRIVER_REMOTE_IDENTIFIER, this.remoteManager.getMyIdentifier())
         .set(EvaluatorShimConfiguration.CONTAINER_IDENTIFIER, containerId)
+        .setMultiple(EvaluatorShimConfiguration.TCP_PORT_SET, this.tcpPortSet)
         .build();
   }
 }
+
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/evaluator/EvaluatorShimConfiguration.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/evaluator/EvaluatorShimConfiguration.java
index c114ba2..be6428e 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/evaluator/EvaluatorShimConfiguration.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/evaluator/EvaluatorShimConfiguration.java
@@ -25,8 +25,14 @@
 import org.apache.reef.runtime.common.launch.REEFMessageCodec;
 import org.apache.reef.tang.formats.ConfigurationModule;
 import org.apache.reef.tang.formats.ConfigurationModuleBuilder;
+import org.apache.reef.tang.formats.OptionalParameter;
 import org.apache.reef.tang.formats.RequiredParameter;
 import org.apache.reef.wake.remote.RemoteConfiguration;
+import org.apache.reef.wake.remote.address.ContainerBasedLocalAddressProvider;
+import org.apache.reef.wake.remote.address.LocalAddressProvider;
+import org.apache.reef.wake.remote.ports.SetTcpPortProvider;
+import org.apache.reef.wake.remote.ports.TcpPortProvider;
+import org.apache.reef.wake.remote.ports.parameters.TcpPortSet;
 
 /**
  * ConfigurationModule to create evaluator shim configurations.
@@ -45,9 +51,28 @@
    */
   public static final RequiredParameter<String> CONTAINER_IDENTIFIER = new RequiredParameter<>();
 
-  public static final ConfigurationModule CONF = new EvaluatorShimConfiguration()
+  /**
+   * Set of TCP Ports.
+   */
+  public static final OptionalParameter<Integer> TCP_PORT_SET = new OptionalParameter<>();
+
+  private static final ConfigurationModule CONF = new EvaluatorShimConfiguration()
       .bindNamedParameter(RemoteConfiguration.MessageCodec.class, REEFMessageCodec.class)
       .bindNamedParameter(DriverRemoteIdentifier.class, DRIVER_REMOTE_IDENTIFIER)
       .bindNamedParameter(ContainerIdentifier.class, CONTAINER_IDENTIFIER)
+      .bindSetEntry(TcpPortSet.class, TCP_PORT_SET)
       .build();
+
+  public static ConfigurationModule getConfigurationModule(final boolean includeContainerConfiguration) {
+    ConfigurationModuleBuilder shimConfigurationBuilder = EvaluatorShimConfiguration.CONF.getBuilder();
+
+    // If using docker containers, then use a different set of bindings
+    if (includeContainerConfiguration) {
+      shimConfigurationBuilder = shimConfigurationBuilder
+          .bindImplementation(LocalAddressProvider.class, ContainerBasedLocalAddressProvider.class)
+          .bindImplementation(TcpPortProvider.class, SetTcpPortProvider.class);
+    }
+
+    return shimConfigurationBuilder.build();
+  }
 }
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerImageName.java
similarity index 71%
copy from lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
copy to lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerImageName.java
index c7f4fbf..3d9ad07 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerImageName.java
@@ -16,22 +16,14 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.reef.wake.remote.ports.parameters;
+package org.apache.reef.runtime.azbatch.parameters;
 
 import org.apache.reef.tang.annotations.Name;
 import org.apache.reef.tang.annotations.NamedParameter;
 
-import java.util.List;
-
 /**
- * An list of tcp port numbers to try.
+ * The container registry image name.
  */
-@NamedParameter(doc = "An list of tcp port numbers to try")
-public final class TcpPortList implements Name<List<Integer>> {
-
-  /**
-   * Empty private constructor to prohibit instantiation of utility class.
-   */
-  private TcpPortList() {
-  }
+@NamedParameter(doc = "The container image name.", default_value = "")
+public final class ContainerImageName implements Name<String> {
 }
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerRegistryPassword.java
similarity index 70%
copy from lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
copy to lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerRegistryPassword.java
index c7f4fbf..acff823 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerRegistryPassword.java
@@ -16,22 +16,17 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.reef.wake.remote.ports.parameters;
+package org.apache.reef.runtime.azbatch.parameters;
 
 import org.apache.reef.tang.annotations.Name;
 import org.apache.reef.tang.annotations.NamedParameter;
 
-import java.util.List;
+// TODO[JIRA REEF-TBD]: Secure the password by encrypting it and using
+// Azure Batch certificates to decrypt on the driver side.
 
 /**
- * An list of tcp port numbers to try.
+ * The container registry password.
  */
-@NamedParameter(doc = "An list of tcp port numbers to try")
-public final class TcpPortList implements Name<List<Integer>> {
-
-  /**
-   * Empty private constructor to prohibit instantiation of utility class.
-   */
-  private TcpPortList() {
-  }
+@NamedParameter(doc = "The container registry password.", default_value = "")
+public final class ContainerRegistryPassword implements Name<String> {
 }
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerRegistryServer.java
similarity index 71%
copy from lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
copy to lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerRegistryServer.java
index c7f4fbf..a001869 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerRegistryServer.java
@@ -16,22 +16,14 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.reef.wake.remote.ports.parameters;
+package org.apache.reef.runtime.azbatch.parameters;
 
 import org.apache.reef.tang.annotations.Name;
 import org.apache.reef.tang.annotations.NamedParameter;
 
-import java.util.List;
-
 /**
- * An list of tcp port numbers to try.
+ * The container registry server name.
  */
-@NamedParameter(doc = "An list of tcp port numbers to try")
-public final class TcpPortList implements Name<List<Integer>> {
-
-  /**
-   * Empty private constructor to prohibit instantiation of utility class.
-   */
-  private TcpPortList() {
-  }
+@NamedParameter(doc = "The container registry server name.", default_value = "")
+public final class ContainerRegistryServer implements Name<String> {
 }
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerRegistryUsername.java
similarity index 71%
copy from lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
copy to lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerRegistryUsername.java
index c7f4fbf..fe6b82e 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/parameters/ContainerRegistryUsername.java
@@ -16,22 +16,14 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.reef.wake.remote.ports.parameters;
+package org.apache.reef.runtime.azbatch.parameters;
 
 import org.apache.reef.tang.annotations.Name;
 import org.apache.reef.tang.annotations.NamedParameter;
 
-import java.util.List;
-
 /**
- * An list of tcp port numbers to try.
+ * The container registry user name.
  */
-@NamedParameter(doc = "An list of tcp port numbers to try")
-public final class TcpPortList implements Name<List<Integer>> {
-
-  /**
-   * Empty private constructor to prohibit instantiation of utility class.
-   */
-  private TcpPortList() {
-  }
+@NamedParameter(doc = "The container registry user name.", default_value = "")
+public final class ContainerRegistryUsername implements Name<String> {
 }
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/batch/AzureBatchHelper.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/batch/AzureBatchHelper.java
index e0753df..f6e59ef 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/batch/AzureBatchHelper.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/batch/AzureBatchHelper.java
@@ -21,18 +21,18 @@
 import com.microsoft.azure.batch.BatchClient;
 import com.microsoft.azure.batch.protocol.models.*;
 
-import org.apache.reef.runtime.azbatch.client.AzureBatchJobSubmissionHandler;
-import org.apache.reef.runtime.azbatch.parameters.AzureBatchPoolId;
+import org.apache.reef.runtime.azbatch.parameters.*;
 import org.apache.reef.runtime.azbatch.util.AzureBatchFileNames;
+import org.apache.reef.runtime.azbatch.util.command.CommandBuilder;
 import org.apache.reef.runtime.azbatch.util.storage.SharedAccessSignatureCloudBlobClientProvider;
 import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.wake.remote.address.ContainerBasedLocalAddressProvider;
+import org.apache.reef.wake.remote.ports.parameters.TcpPortSet;
 
 import javax.inject.Inject;
 import java.io.IOException;
 import java.net.URI;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -41,27 +41,39 @@
  */
 public final class AzureBatchHelper {
 
-  private static final Logger LOG = Logger.getLogger(AzureBatchJobSubmissionHandler.class.getName());
+  private static final Logger LOG = Logger.getLogger(AzureBatchHelper.class.getName());
 
   /*
    * Environment variable that contains the Azure Batch jobId.
    */
   private static final String AZ_BATCH_JOB_ID_ENV = "AZ_BATCH_JOB_ID";
+  private static final String CAPTURE_HOST_IP_ADDRESS_TASK_NAME = "CaptureHostIpAddress";
 
   private final AzureBatchFileNames azureBatchFileNames;
 
   private final BatchClient client;
   private final PoolInformation poolInfo;
+  private final CommandBuilder commandBuilder;
+  private final ContainerRegistryProvider containerRegistryProvider;
+  private final boolean areContainersEnabled;
+  private final Set<Integer> ports;
 
   @Inject
   public AzureBatchHelper(
       final AzureBatchFileNames azureBatchFileNames,
       final IAzureBatchCredentialProvider credentialProvider,
+      final CommandBuilder commandBuilder,
+      final ContainerRegistryProvider containerRegistryProvider,
+      @Parameter(TcpPortSet.class) final Set<Integer> ports,
       @Parameter(AzureBatchPoolId.class) final String azureBatchPoolId) {
     this.azureBatchFileNames = azureBatchFileNames;
 
     this.client = BatchClient.open(credentialProvider.getCredentials());
     this.poolInfo = new PoolInformation().withPoolId(azureBatchPoolId);
+    this.commandBuilder = commandBuilder;
+    this.containerRegistryProvider = containerRegistryProvider;
+    this.ports = ports;
+    this.areContainersEnabled = this.containerRegistryProvider.isValid();
   }
 
   /**
@@ -75,7 +87,7 @@
    */
   public void submitJob(final String applicationId, final String storageContainerSAS, final URI jobJarUri,
                         final String command) throws IOException {
-    ResourceFile jarResourceFile = new ResourceFile()
+    final ResourceFile jarResourceFile = new ResourceFile()
         .withBlobSource(jobJarUri.toString())
         .withFilePath(AzureBatchFileNames.getTaskJarFileName());
 
@@ -83,26 +95,29 @@
     // as an environment variable.
     // See https://docs.microsoft.com/en-us/dotnet/api/microsoft.azure.batch.cloudtask.authenticationtokensettings
     // for more info.
-    AuthenticationTokenSettings authenticationTokenSettings = new AuthenticationTokenSettings();
-    authenticationTokenSettings.withAccess(Collections.singletonList(AccessScope.JOB));
+    final AuthenticationTokenSettings authenticationTokenSettings = new AuthenticationTokenSettings()
+        .withAccess(Collections.singletonList(AccessScope.JOB));
 
-    EnvironmentSetting environmentSetting = new EnvironmentSetting()
+    final EnvironmentSetting environmentSetting = new EnvironmentSetting()
         .withName(SharedAccessSignatureCloudBlobClientProvider.AZURE_STORAGE_CONTAINER_SAS_TOKEN_ENV)
         .withValue(storageContainerSAS);
 
-    JobManagerTask jobManagerTask = new JobManagerTask()
+    final JobManagerTask jobManagerTask = new JobManagerTask()
         .withRunExclusive(false)
         .withId(applicationId)
         .withResourceFiles(Collections.singletonList(jarResourceFile))
         .withEnvironmentSettings(Collections.singletonList(environmentSetting))
         .withAuthenticationTokenSettings(authenticationTokenSettings)
+        .withKillJobOnCompletion(true)
+        .withContainerSettings(createTaskContainerSettings(applicationId))
         .withCommandLine(command);
 
-    LOG.log(Level.INFO, "Job Manager (aka driver) task command: " + command);
+    LOG.log(Level.INFO, "Job Manager (aka driver) task command: {0}", command);
 
-    JobAddParameter jobAddParameter = new JobAddParameter()
+    final JobAddParameter jobAddParameter = new JobAddParameter()
         .withId(applicationId)
         .withJobManagerTask(jobManagerTask)
+        .withJobPreparationTask(createJobPreparationTask())
         .withPoolInfo(poolInfo);
 
     client.jobOperations().createJob(jobAddParameter);
@@ -134,13 +149,19 @@
         .withFilePath(this.azureBatchFileNames.getEvaluatorShimConfigurationPath());
     resources.add(confSourceFile);
 
-    LOG.log(Level.INFO, "Evaluator task command: " + command);
+    LOG.log(Level.INFO, "Evaluator task command: {0}", command);
 
-    final TaskAddParameter taskAddParameter = new TaskAddParameter()
+    TaskAddParameter taskAddParameter = new TaskAddParameter()
         .withId(taskId)
         .withResourceFiles(resources)
+        .withContainerSettings(createTaskContainerSettings(taskId))
         .withCommandLine(command);
 
+    if (this.areContainersEnabled) {
+      taskAddParameter = taskAddParameter.withUserIdentity(
+          new UserIdentity().withAutoUser(new AutoUserSpecification().withElevationLevel(ElevationLevel.ADMIN)));
+    }
+
     this.client.taskOperations().createTask(jobId, taskAddParameter);
   }
 
@@ -169,4 +190,44 @@
   public String getAzureBatchJobId() {
     return System.getenv(AZ_BATCH_JOB_ID_ENV);
   }
+
+  private TaskContainerSettings createTaskContainerSettings(final String dockerContainerId) {
+    if (!this.areContainersEnabled) {
+      return null;
+    }
+
+    final StringBuilder runOptions = new StringBuilder(String.format(
+        "-d --rm --name %s --env %s=%s ",
+        dockerContainerId,
+        ContainerBasedLocalAddressProvider.HOST_IP_ADDR_PATH_ENV,
+        this.commandBuilder.getIpAddressFilePath()));
+
+    for (final int port : this.ports) {
+      runOptions.append(String.format("-p %d:%d ", port, port));
+    }
+
+    return new TaskContainerSettings()
+        .withRegistry(this.containerRegistryProvider.getContainerRegistry())
+        .withImageName(this.containerRegistryProvider.getContainerImageName())
+        .withContainerRunOptions(runOptions.toString());
+  }
+
+  private JobPreparationTask createJobPreparationTask() {
+    if (!this.areContainersEnabled) {
+      return null;
+    }
+
+    final String captureIpAddressCommandLine = this.commandBuilder.captureIpAddressCommandLine();
+    return new JobPreparationTask()
+        .withId(CAPTURE_HOST_IP_ADDRESS_TASK_NAME)
+        .withCommandLine(captureIpAddressCommandLine);
+  }
+
+  public static Set<String> toStringSet(final Collection<?> items) {
+    final Set<String> set = new HashSet<>(items.size());
+    for (final Object elem: items) {
+      set.add(elem.toString());
+    }
+    return Collections.unmodifiableSet(set);
+  }
 }
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/batch/ContainerRegistryProvider.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/batch/ContainerRegistryProvider.java
new file mode 100644
index 0000000..1bfafa9
--- /dev/null
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/batch/ContainerRegistryProvider.java
@@ -0,0 +1,84 @@
+/*
+ * 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.reef.runtime.azbatch.util.batch;
+
+import com.microsoft.azure.batch.protocol.models.ContainerRegistry;
+import org.apache.commons.lang.StringUtils;
+import org.apache.reef.runtime.azbatch.parameters.ContainerImageName;
+import org.apache.reef.runtime.azbatch.parameters.ContainerRegistryPassword;
+import org.apache.reef.runtime.azbatch.parameters.ContainerRegistryServer;
+import org.apache.reef.runtime.azbatch.parameters.ContainerRegistryUsername;
+import org.apache.reef.tang.annotations.Parameter;
+
+import javax.inject.Inject;
+
+/**
+ * Class that holds the settings of the Azure container service settings.
+ */
+public final class ContainerRegistryProvider {
+
+  private final String containerRegistryServer;
+  private final String containerRegistryUsername;
+  private final String containerRegistryPassword;
+  private final String containerImageName;
+
+  @Inject
+  private ContainerRegistryProvider(
+      @Parameter(ContainerRegistryServer.class) final String containerRegistryServer,
+      @Parameter(ContainerRegistryUsername.class) final String containerRegistryUsername,
+      @Parameter(ContainerRegistryPassword.class) final String containerRegistryPassword,
+      @Parameter(ContainerImageName.class) final String containerImageName
+  ) {
+    this.containerRegistryServer = containerRegistryServer;
+    this.containerRegistryUsername = containerRegistryUsername;
+    this.containerRegistryPassword = containerRegistryPassword;
+    this.containerImageName = containerImageName;
+  }
+
+  public boolean isValid() {
+    return !StringUtils.isEmpty(this.containerRegistryServer);
+  }
+
+  public String getContainerRegistryServer() {
+    return this.containerRegistryServer;
+  }
+
+  public String getContainerRegistryUsername() {
+    return this.containerRegistryUsername;
+  }
+
+  public String getContainerRegistryPassword() {
+    return this.containerRegistryPassword;
+  }
+
+  public String getContainerImageName() {
+    return this.containerImageName;
+  }
+
+  public ContainerRegistry getContainerRegistry() {
+    if (!this.isValid()) {
+      throw new RuntimeException("Container registry is invalid");
+    }
+
+    return new ContainerRegistry()
+        .withRegistryServer(this.getContainerRegistryServer())
+        .withUserName(this.getContainerRegistryUsername())
+        .withPassword(this.getContainerRegistryPassword());
+  }
+}
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/CommandBuilder.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/CommandBuilder.java
index 96cd716..b05e010 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/CommandBuilder.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/CommandBuilder.java
@@ -53,4 +53,18 @@
    */
   String buildEvaluatorCommand(final ResourceLaunchEvent resourceLaunchEvent,
                                final int containerMemory, final double jvmHeapFactor);
+
+  /**
+   * Returns the path to a file where the ip address is persisted.
+   *
+   * @return path to the file.
+   */
+  String getIpAddressFilePath();
+
+  /**
+   * Returns a command line that saves the ip address of the host to a file.
+   *
+   * @return command line string.
+   */
+  String captureIpAddressCommandLine();
 }
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/LinuxCommandBuilder.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/LinuxCommandBuilder.java
index adb58f5..0865662 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/LinuxCommandBuilder.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/LinuxCommandBuilder.java
@@ -65,4 +65,15 @@
   protected String getEvaluatorShimClasspath() {
     return StringUtils.join(super.classpathProvider.getEvaluatorClasspath(), CLASSPATH_SEPARATOR_CHAR);
   }
+
+  @Override
+  public String getIpAddressFilePath() {
+    return "$AZ_BATCH_JOB_PREP_WORKING_DIR/hostip.txt";
+  }
+
+  @Override
+  public String captureIpAddressCommandLine() {
+    final String filePath = getIpAddressFilePath();
+    return String.format("/bin/bash -c \"rm -f %s; echo `hostname -i` > %s\"", filePath, filePath);
+  }
 }
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/WindowsCommandBuilder.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/WindowsCommandBuilder.java
index f382bef..ea0e097 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/WindowsCommandBuilder.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/command/WindowsCommandBuilder.java
@@ -71,4 +71,16 @@
     return String.format("'%s'", StringUtils.join(
         super.classpathProvider.getEvaluatorClasspath(), CLASSPATH_SEPARATOR_CHAR));
   }
+
+  @Override
+  public String getIpAddressFilePath() {
+    return "%AZ_BATCH_JOB_PREP_WORKING_DIR%\\hostip.txt";
+  }
+
+  @Override
+  public String captureIpAddressCommandLine() {
+    return String.format("powershell /c \"Set-Content -Path %s -Value "
+        + "((Test-Connection -ComputerName $Env:ComputerName -Count 1).IPV4Address.IPAddressToString) "
+        + " -NoNewline -Force\"", getIpAddressFilePath());
+  }
 }
diff --git a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/storage/SharedAccessSignatureCloudBlobClientProvider.java b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/storage/SharedAccessSignatureCloudBlobClientProvider.java
index e678445..a087488 100644
--- a/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/storage/SharedAccessSignatureCloudBlobClientProvider.java
+++ b/lang/java/reef-runtime-azbatch/src/main/java/org/apache/reef/runtime/azbatch/util/storage/SharedAccessSignatureCloudBlobClientProvider.java
@@ -100,7 +100,7 @@
       }
 
       URI result = builder.addToURI(uri);
-      LOG.log(Level.INFO, "Here's the URI: " + result);
+      LOG.log(Level.INFO, "Here's the URI: {0}", result);
 
       return result;
     } catch (StorageException | URISyntaxException e) {
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/address/ContainerBasedLocalAddressProvider.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/address/ContainerBasedLocalAddressProvider.java
new file mode 100644
index 0000000..2b63935
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/address/ContainerBasedLocalAddressProvider.java
@@ -0,0 +1,113 @@
+/*
+ * 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.reef.wake.remote.address;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.Tang;
+
+import javax.inject.Inject;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * A LocalAddressProvider that reads the contents of the file at HOST_IP_ADDR_PATH and uses it to be the ip address.
+ */
+public final class ContainerBasedLocalAddressProvider implements LocalAddressProvider {
+
+  private static final Pattern IPADDRESS_PATTERN = Pattern.compile(
+      "(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)");
+
+  public static final String HOST_IP_ADDR_PATH_ENV = "HOST_IP_ADDR_PATH";
+  private static final Logger LOG = Logger.getLogger(ContainerBasedLocalAddressProvider.class.getName());
+
+  private String cachedLocalAddress = null;
+
+  /**
+   * The constructor is for Tang only.
+   */
+  @Inject
+  private ContainerBasedLocalAddressProvider() {
+    LOG.log(Level.FINE, "Instantiating ContainerBasedLocalAddressProvider");
+  }
+
+  @Override
+  public synchronized String getLocalAddress() {
+    if (cachedLocalAddress != null) {
+      return cachedLocalAddress;
+    }
+
+    final String ipAddressPath = System.getenv(HOST_IP_ADDR_PATH_ENV);
+    LOG.log(Level.FINE, "IpAddressPath is {0}", ipAddressPath);
+    if (StringUtils.isEmpty(ipAddressPath)) {
+      final String message = String.format("Environment variable must be set for %s", HOST_IP_ADDR_PATH_ENV);
+      LOG.log(Level.SEVERE, message);
+      throw new RuntimeException(message);
+    }
+
+    final File ipAddressFile = new File(ipAddressPath);
+    if (!ipAddressFile.exists() || !ipAddressFile.isFile()) {
+      final String message = String.format("%s points to invalid path: %s", HOST_IP_ADDR_PATH_ENV, ipAddressPath);
+      LOG.log(Level.SEVERE, message);
+      throw new RuntimeException(message);
+    }
+
+    try {
+      cachedLocalAddress = readFile(ipAddressPath, StandardCharsets.UTF_8);
+      return cachedLocalAddress;
+    } catch (IOException e) {
+      final String message = String.format("Exception when attempting to read file %s", ipAddressPath);
+      LOG.log(Level.SEVERE, message, e);
+      throw new RuntimeException(message, e);
+    }
+  }
+
+  @Override
+  public Configuration getConfiguration() {
+    return Tang.Factory.getTang().newConfigurationBuilder()
+        .bind(LocalAddressProvider.class, ContainerBasedLocalAddressProvider.class)
+        .build();
+  }
+
+  @Override
+  public String toString() {
+    return "ContainerBasedLocalAddressProvider:" + this.getLocalAddress();
+  }
+
+  private String readFile(final String path, final Charset encoding)
+      throws IOException {
+    final byte[] encoded = Files.readAllBytes(Paths.get(path));
+    final String ipString = new String(encoded, encoding);
+    final Matcher matcher = IPADDRESS_PATTERN.matcher(StringUtils.trim(ipString));
+
+    if (!matcher.matches()) {
+      throw new RuntimeException(String.format("File at location %s has invalid ip address", path));
+    }
+
+    return matcher.group();
+  }
+}
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/impl/DefaultRemoteManagerImplementation.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/impl/DefaultRemoteManagerImplementation.java
index 4566aca..2e4f3d3 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/impl/DefaultRemoteManagerImplementation.java
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/impl/DefaultRemoteManagerImplementation.java
@@ -27,7 +27,6 @@
 import org.apache.reef.wake.remote.ports.TcpPortProvider;
 import org.apache.reef.wake.remote.transport.Transport;
 import org.apache.reef.wake.remote.transport.TransportFactory;
-import org.apache.reef.wake.remote.transport.netty.NettyMessagingTransport;
 
 import javax.inject.Inject;
 import java.net.InetSocketAddress;
@@ -53,11 +52,6 @@
    */
   private static final long CLOSE_EXECUTOR_TIMEOUT = 10000; //ms
 
-  /**
-   * Indicates a hostname that isn't set or known.
-   */
-  public static final String UNKNOWN_HOST_NAME = NettyMessagingTransport.UNKNOWN_HOST_NAME;
-
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private final RemoteSeqNumGenerator seqGen = new RemoteSeqNumGenerator();
 
@@ -95,7 +89,10 @@
 
     this.handlerContainer.setTransport(this.transport);
 
-    this.myIdentifier = new SocketRemoteIdentifier((InetSocketAddress)this.transport.getLocalAddress());
+    InetSocketAddress address = new InetSocketAddress(
+        localAddressProvider.getLocalAddress(),
+        this.transport.getListeningPort());
+    this.myIdentifier = new SocketRemoteIdentifier(address);
 
     this.reSendStage = new RemoteSenderStage(codec, this.transport, 10);
 
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/ListTcpPortProvider.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/SetTcpPortProvider.java
similarity index 69%
rename from lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/ListTcpPortProvider.java
rename to lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/SetTcpPortProvider.java
index c28a638..a6108af 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/ListTcpPortProvider.java
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/SetTcpPortProvider.java
@@ -18,28 +18,27 @@
  */
 package org.apache.reef.wake.remote.ports;
 
-
 import org.apache.commons.lang.StringUtils;
 import org.apache.reef.tang.annotations.Parameter;
-import org.apache.reef.wake.remote.ports.parameters.TcpPortList;
+import org.apache.reef.wake.remote.ports.parameters.TcpPortSet;
 
 import javax.inject.Inject;
 import java.util.Iterator;
-import java.util.List;
+import java.util.Set;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
 /**
- * A TcpPortProvider which gives out random ports in a range.
+ * A TcpPortProvider which gives out ports in a set.
  */
-public final class ListTcpPortProvider implements TcpPortProvider {
+public final class SetTcpPortProvider implements TcpPortProvider {
 
-  private static final Logger LOG = Logger.getLogger(ListTcpPortProvider.class.getName());
-  private final List<Integer> tcpPortList;
+  private static final Logger LOG = Logger.getLogger(SetTcpPortProvider.class.getName());
+  private final Set<Integer> tcpPortSet;
 
   @Inject
-  public ListTcpPortProvider(@Parameter(TcpPortList.class) final List<Integer> tcpPortList) {
-    this.tcpPortList = tcpPortList;
+  public SetTcpPortProvider(@Parameter(TcpPortSet.class) final Set<Integer> tcpPortSet) {
+    this.tcpPortSet = tcpPortSet;
     LOG.log(Level.FINE, "Instantiating {0}", this);
   }
 
@@ -50,11 +49,11 @@
    */
   @Override
   public Iterator<Integer> iterator() {
-    return this.tcpPortList.iterator();
+    return this.tcpPortSet.iterator();
   }
 
   @Override
   public String toString() {
-    return "ListTcpPortProvider{" + StringUtils.join(this.tcpPortList, ',') + '}';
+    return "SetTcpPortProvider{" + StringUtils.join(this.tcpPortSet, ',') + '}';
   }
 }
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortSet.java
similarity index 83%
rename from lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
rename to lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortSet.java
index c7f4fbf..47025e8 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortList.java
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/ports/parameters/TcpPortSet.java
@@ -20,18 +20,17 @@
 
 import org.apache.reef.tang.annotations.Name;
 import org.apache.reef.tang.annotations.NamedParameter;
-
-import java.util.List;
+import java.util.Set;
 
 /**
- * An list of tcp port numbers to try.
+ * A set of tcp port numbers to try.
  */
-@NamedParameter(doc = "An list of tcp port numbers to try")
-public final class TcpPortList implements Name<List<Integer>> {
+@NamedParameter(doc = "Set of tcp port numbers")
+public final class TcpPortSet implements Name<Set<Integer>> {
 
   /**
    * Empty private constructor to prohibit instantiation of utility class.
    */
-  private TcpPortList() {
+  private TcpPortSet() {
   }
 }
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyMessagingTransport.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyMessagingTransport.java
index ea221ec..b51905e 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyMessagingTransport.java
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/remote/transport/netty/NettyMessagingTransport.java
@@ -255,9 +255,9 @@
           return link;
         }
       }
-      
+
       if (i == this.numberOfTries) {
-        // Connection failure 
+        // Connection failure
         throw new ConnectException("Connection to " + remoteAddr + " refused");
       }
 
@@ -323,7 +323,7 @@
         }
       }
     }
-    
+
     return link;
   }
 
diff --git a/lang/java/reef-webserver/src/main/java/org/apache/reef/webserver/HttpServerImpl.java b/lang/java/reef-webserver/src/main/java/org/apache/reef/webserver/HttpServerImpl.java
index 67dce8b..a10a4eb 100644
--- a/lang/java/reef-webserver/src/main/java/org/apache/reef/webserver/HttpServerImpl.java
+++ b/lang/java/reef-webserver/src/main/java/org/apache/reef/webserver/HttpServerImpl.java
@@ -21,16 +21,15 @@
 import org.apache.reef.tang.annotations.Parameter;
 import org.apache.reef.util.logging.LoggingScope;
 import org.apache.reef.util.logging.LoggingScopeFactory;
+import org.apache.reef.wake.remote.RemoteConfiguration;
 import org.apache.reef.wake.remote.address.LocalAddressProvider;
 import org.apache.reef.wake.remote.ports.TcpPortProvider;
-import org.apache.reef.wake.remote.ports.parameters.TcpPortRangeBegin;
 import org.mortbay.jetty.Connector;
 import org.mortbay.jetty.Server;
 import org.mortbay.jetty.bio.SocketConnector;
 
 import javax.inject.Inject;
 import java.net.BindException;
-import java.util.Iterator;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -39,6 +38,11 @@
  */
 public final class HttpServerImpl implements HttpServer {
   /**
+   * Indicates a hostname that isn't set or known.
+   */
+  private static final String UNKNOWN_HOST_NAME = "##UNKNOWN##";
+
+  /**
    * Standard Java logger.
    */
   private static final Logger LOG = Logger.getLogger(HttpServerImpl.class.getName());
@@ -64,35 +68,36 @@
   private final LoggingScopeFactory loggingScopeFactory;
 
   /**
-   * The address provider for the HTTPServer.
+   * The host address for the HTTPServer.
    */
-  private final LocalAddressProvider addressProvider;
+  private final String hostAddress;
 
   /**
    * Constructor of HttpServer that wraps Jetty Server.
    *
    * @param jettyHandler
-   * @param portNumber
    * @throws Exception
    */
   @Inject
-  HttpServerImpl(final JettyHandler jettyHandler,
+  HttpServerImpl(@Parameter(RemoteConfiguration.HostAddress.class) final String hostAddress,
+                 final JettyHandler jettyHandler,
                  final LocalAddressProvider addressProvider,
-                 @Parameter(TcpPortRangeBegin.class)final int portNumber,
                  final TcpPortProvider tcpPortProvider,
                  final LoggingScopeFactory loggingScopeFactory) throws Exception {
-    this.addressProvider = addressProvider;
     this.loggingScopeFactory = loggingScopeFactory;
     this.jettyHandler = jettyHandler;
-    int availablePort = portNumber;
-    Server srv = null;
 
+    this.hostAddress = UNKNOWN_HOST_NAME.equals(hostAddress) ? addressProvider.getLocalAddress() : hostAddress;
     try (final LoggingScope ls = this.loggingScopeFactory.httpServer()) {
 
-      final Iterator<Integer> ports = tcpPortProvider.iterator();
-      while (ports.hasNext() && srv  == null) {
-        availablePort = ports.next();
-        srv = tryPort(availablePort);
+      Server srv = null;
+      int availablePort = -1;
+      for (final int p : tcpPortProvider) {
+        srv = tryPort(p);
+        if (srv != null) {
+          availablePort = p;
+          break;
+        }
       }
 
       if (srv  != null) {
@@ -109,7 +114,7 @@
   private Server tryPort(final int portNumber) throws Exception {
     Server srv = new Server();
     final Connector connector = new SocketConnector();
-    connector.setHost(addressProvider.getLocalAddress());
+    connector.setHost(this.hostAddress);
     connector.setPort(portNumber);
     srv.addConnector(connector);
     try {
@@ -117,7 +122,8 @@
       LOG.log(Level.INFO, "Jetty Server started with port: {0}", portNumber);
     } catch (final BindException ex) {
       srv = null;
-      LOG.log(Level.FINEST, "Cannot use port: {0}. Will try another", portNumber);
+      LOG.log(Level.FINEST, "Cannot use host: {0},port: {1}. Will try another",
+          new Object[] {this.hostAddress, portNumber});
     }
     return srv;
   }