class LocalBaseImageSteps {

  @VisibleForTesting
  static LocalImage cacheDockerImageTar(
      BuildContext buildContext,
      Path tarPath,
      ProgressEventDispatcher.Factory progressEventDispatcherFactory,
      TempDirectoryProvider tempDirectoryProvider)
      throws IOException, LayerCountMismatchException {
    ExecutorService executorService = buildContext.getExecutorService();
    Path destination = tempDirectoryProvider.newDirectory();

    try (TimerEventDispatcher ignored =
        new TimerEventDispatcher(
            buildContext.getEventHandlers(),
            "Extracting tar " + tarPath + " into " + destination)) {
      TarExtractor.extract(tarPath, destination);

      InputStream manifestStream = Files.newInputStream(destination.resolve("manifest.json"));
      DockerManifestEntryTemplate loadManifest =
          new ObjectMapper()
              .configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES, true)
              .readValue(manifestStream, DockerManifestEntryTemplate[].class)[0];
      manifestStream.close();

      Path configPath = destination.resolve(loadManifest.getConfig());
      ContainerConfigurationTemplate configurationTemplate =
          JsonTemplateMapper.readJsonFromFile(configPath, ContainerConfigurationTemplate.class);
      // Don't compute the digest of the loaded Java JSON instance.
      BlobDescriptor originalConfigDescriptor =
          Blobs.from(configPath).writeTo(ByteStreams.nullOutputStream());

      List<String> layerFiles = loadManifest.getLayerFiles();
      if (configurationTemplate.getLayerCount() != layerFiles.size()) {
        throw new LayerCountMismatchException(
            "Invalid base image format: manifest contains "
                + layerFiles.size()
                + " layers, but container configuration contains "
                + configurationTemplate.getLayerCount()
                + " layers");
      }
      buildContext
          .getBaseImageLayersCache()
          .writeLocalConfig(originalConfigDescriptor.getDigest(), configurationTemplate);

      // Check the first layer to see if the layers are compressed already. 'docker save' output
      // is uncompressed, but a jib-built tar has compressed layers.
      boolean layersAreCompressed =
          layerFiles.size() > 0 && isGzipped(destination.resolve(layerFiles.get(0)));

      // Process layer blobs
      try (ProgressEventDispatcher progressEventDispatcher =
          progressEventDispatcherFactory.create(
              "processing base image layers", layerFiles.size())) {
        // Start compressing layers in parallel
        List<Future<PreparedLayer>> preparedLayers = new ArrayList<>();
        for (int index = 0; index < layerFiles.size(); index++) {
          Path layerFile = destination.resolve(layerFiles.get(index));
          DescriptorDigest diffId = configurationTemplate.getLayerDiffId(index);
          ProgressEventDispatcher.Factory layerProgressDispatcherFactory =
              progressEventDispatcher.newChildProducer();
          preparedLayers.add(
              executorService.submit(
                  () ->
                      compressAndCacheTarLayer(
                          buildContext.getBaseImageLayersCache(),
                          diffId,
                          layerFile,
                          layersAreCompressed,
                          layerProgressDispatcherFactory)));
        }
        return new LocalImage(preparedLayers, configurationTemplate);
      }
    }
  }

}