diff --git a/pkg/transformer/kubernetes/kubernetes.go b/pkg/transformer/kubernetes/kubernetes.go
index c674b34d..f26689ba 100644
--- a/pkg/transformer/kubernetes/kubernetes.go
+++ b/pkg/transformer/kubernetes/kubernetes.go
@@ -81,6 +81,8 @@ const (
 	DaemonSetController = "daemonset"
 	// StatefulStateController is controller type for StatefulSet
 	StatefulStateController = "statefulset"
+	// JobController is controller type for Job
+	JobController = "job"
 )
 
 // CheckUnsupportedKey checks if given komposeObject contains
@@ -523,6 +525,52 @@ func (k *Kubernetes) InitDS(name string, service kobject.ServiceConfig, opt kobj
 	return ds
 }
 
+// InitJob initializes Kubernetes Job object
+func (k *Kubernetes) InitJob(name string, service kobject.ServiceConfig, opt kobject.ConvertOptions) *batchv1.Job {
+	var podSpec api.PodSpec
+	if len(service.Configs) > 0 {
+		podSpec = k.InitPodSpecWithConfigMap(name, service, opt)
+	} else {
+		podSpec = k.InitPodSpec(name, service, opt)
+	}
+
+	// Jobs need RestartPolicy set (default is Always which is invalid for Jobs)
+	// Use Never for jobs unless explicitly set
+	if podSpec.RestartPolicy == "" || podSpec.RestartPolicy == api.RestartPolicyAlways {
+		podSpec.RestartPolicy = api.RestartPolicyNever
+	}
+
+	job := &batchv1.Job{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Job",
+			APIVersion: "batch/v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   name,
+			Labels: transformer.ConfigAllLabels(name, &service),
+		},
+		Spec: batchv1.JobSpec{
+			Template: api.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: transformer.ConfigLabels(name),
+				},
+				Spec: podSpec,
+			},
+		},
+	}
+
+	// Set backoffLimit from label if specified
+	if backoffLimit, ok := service.Labels["kompose.job.backoff-limit"]; ok {
+		limit, err := strconv.ParseInt(backoffLimit, 10, 32)
+		if err == nil {
+			limit32 := int32(limit)
+			job.Spec.BackoffLimit = &limit32
+		}
+	}
+
+	return job
+}
+
 // InitSS method initialize a stateful set
 func (k *Kubernetes) InitSS(name string, service kobject.ServiceConfig, replicas int, opt kobject.ConvertOptions) *appsv1.StatefulSet {
 	var podSpec api.PodSpec
@@ -1454,6 +1502,10 @@ func (k *Kubernetes) CreateWorkloadAndConfigMapObjects(name string, service kobj
 		objects = append(objects, k.InitSS(name, service, replica, opt))
 	}
 
+	if opt.Controller == JobController {
+		objects = append(objects, k.InitJob(name, service, opt))
+	}
+
 	envConfigMaps := k.PargeEnvFiletoConfigMaps(name, service, opt)
 	objects = append(objects, envConfigMaps...)
 	return objects
@@ -1807,7 +1859,10 @@ func (k *Kubernetes) Transform(komposeObject kobject.KomposeObject, opt kobject.
 		}
 
 		// Generate pod or cronjob and configmap objects
-		if (service.Restart == "no" || service.Restart == "on-failure") && !opt.IsPodController() {
+		// Check if service has explicit controller type label
+		_, hasControllerLabel := service.Labels[compose.LabelControllerType]
+
+		if (service.Restart == "no" || service.Restart == "on-failure") && !opt.IsPodController() && !hasControllerLabel {
 			if service.CronJobSchedule != "" {
 				log.Infof("Create kubernetes pod instead of pod controller due to restart policy: %s", service.Restart)
 				cronJob := k.InitCJ(name, service, service.CronJobSchedule, service.CronJobConcurrencyPolicy, service.CronJobBackoffLimit, opt)
@@ -1881,6 +1936,12 @@ func (k *Kubernetes) UpdateController(obj runtime.Object, updateTemplate func(*a
 			return errors.Wrap(err, "updateTemplate failed")
 		}
 		updateMeta(&t.ObjectMeta)
+	case *batchv1.Job:
+		err = updateTemplate(&t.Spec.Template)
+		if err != nil {
+			return errors.Wrap(err, "updateTemplate failed")
+		}
+		updateMeta(&t.ObjectMeta)
 	case *deployapi.DeploymentConfig:
 		err = updateTemplate(t.Spec.Template)
 		if err != nil {