class FileSecretApplier extends KubernetesSecretApplier<KubernetesEnvironment> {

  @Override
  public void applySecret(KubernetesEnvironment env, RuntimeIdentity runtimeIdentity, Secret secret)
      throws InfrastructureException {
    final String secretMountPath = secret.getMetadata().getAnnotations().get(ANNOTATION_MOUNT_PATH);
    boolean secretAutomount =
        Boolean.parseBoolean(secret.getMetadata().getAnnotations().get(ANNOTATION_AUTOMOUNT));
    if (secretMountPath == null) {
      throw new InfrastructureException(
          format(
              "Unable to mount secret '%s': It is configured to be mounted as a file but the mount path was not specified. Please define the '%s' annotation on the secret to specify it.",
              secret.getMetadata().getName(), ANNOTATION_MOUNT_PATH));
    }

    Volume volumeFromSecret =
        new VolumeBuilder()
            .withName(secret.getMetadata().getName())
            .withSecret(
                new SecretVolumeSourceBuilder()
                    .withNewSecretName(secret.getMetadata().getName())
                    .build())
            .build();

    for (PodData podData : env.getPodsData().values()) {
      if (!podData.getRole().equals(PodRole.DEPLOYMENT)) {
        continue;
      }
      if (podData
          .getSpec()
          .getVolumes()
          .stream()
          .anyMatch(v -> v.getName().equals(volumeFromSecret.getName()))) {
        volumeFromSecret.setName(volumeFromSecret.getName() + "_" + NameGenerator.generate("", 6));
      }

      podData.getSpec().getVolumes().add(volumeFromSecret);

      for (Container container : podData.getSpec().getContainers()) {
        Optional<ComponentImpl> component = getComponent(env, container.getName());
        // skip components that explicitly disable automount
        if (component.isPresent() && isComponentAutomountFalse(component.get())) {
          continue;
        }
        // if automount disabled globally and not overridden in component
        if (!secretAutomount
            && (!component.isPresent() || !isComponentAutomountTrue(component.get()))) {
          continue;
        }
        // find path override if any
        Optional<String> overridePathOptional = Optional.empty();
        if (component.isPresent()) {
          overridePathOptional =
              getOverridenComponentPath(component.get(), secret.getMetadata().getName());
        }
        final String componentMountPath = overridePathOptional.orElse(secretMountPath);
        // it's not possible to mount multiple volumes on same path on older k8s, so we
        // remove the existing mount here to replace it with new one.
        if (k8sVersion.olderThan(1, 13)) {
          LOG.debug(
              "Unable to mount multiple VolumeMounts on same path on this k8s version. Removing conflicting volumes in favor of secret mounts.");
          container
              .getVolumeMounts()
              .removeIf(vm -> Paths.get(vm.getMountPath()).equals(Paths.get(componentMountPath)));
        }

        container
            .getVolumeMounts()
            .addAll(
                secret
                    .getData()
                    .keySet()
                    .stream()
                    .map(
                        secretFile ->
                            buildVolumeMount(volumeFromSecret, componentMountPath, secretFile))
                    .collect(Collectors.toList()));
      }
    }
  }

  @Inject public  FileSecretApplier(K8sVersion k8sVersion);

}