public void clearStreamsFromPreviousRun(Config prevConfig) {
    try {
      ApplicationConfig appConfig = new ApplicationConfig(prevConfig);
      LOGGER.info("run.id from previous run is {}", appConfig.getRunId());

      StreamConfig streamConfig = new StreamConfig(prevConfig);

      //Find all intermediate streams and clean up
      Set<StreamSpec> intStreams = streamConfig.getStreamIds().stream()
          .filter(streamConfig::getIsIntermediateStream)
          .map(id -> new StreamSpec(id, streamConfig.getPhysicalName(id), streamConfig.getSystem(id)))
          .collect(Collectors.toSet());
      intStreams.forEach(stream -> {
        LOGGER.info("Clear intermediate stream {} in system {}", stream.getPhysicalName(), stream.getSystemName());
        systemAdmins.getSystemAdmin(stream.getSystemName()).clearStream(stream);
      });

      //Find checkpoint stream and clean up
      TaskConfig taskConfig = new TaskConfig(prevConfig);
      taskConfig.getCheckpointManager(new MetricsRegistryMap()).ifPresent(CheckpointManager::clearCheckpoints);

      //Find changelog streams and remove them
      StorageConfig storageConfig = new StorageConfig(prevConfig);
      for (String store : storageConfig.getStoreNames()) {
        String changelog = storageConfig.getChangelogStream(store).orElse(null);
        if (changelog != null) {
          LOGGER.info("Clear store {} changelog {}", store, changelog);
          SystemStream systemStream = StreamUtil.getSystemStreamFromNames(changelog);
          StreamSpec spec = StreamSpec.createChangeLogStreamSpec(systemStream.getStream(), systemStream.getSystem(), 1);
          systemAdmins.getSystemAdmin(spec.getSystemName()).clearStream(spec);
        }
      }
    } catch (Exception e) {
      // For batch, we always create a new set of internal streams (checkpoint, changelog and intermediate) with unique
      // id. So if clearStream doesn't work, it won't affect the correctness of the results.
      // We log a warning here and rely on retention to clean up the streams later.
      LOGGER.warn("Fail to clear internal streams from previous run. Please clean up manually.", e);
    }
  }
This is a best-effort approach to clear the internal streams from previous run, including intermediate streams,
checkpoint stream and changelog streams.
For batch processing, we always clean up the previous internal streams and create a new set for each run.
@param prevConfig config of the previous run
samza-core/src/main/java/org/apache/samza/execution/StreamManager.java