Marked as final to prevent override. Subclasses should implement the doPrepare() method.
@param conf
@param topologyContext
@param collector
5309
public final void prepare(Map conf, TopologyContext topologyContext, OutputCollector collector){
        this.writeLock = new Object();
        if (this.syncPolicy == null) throw new IllegalStateException("SyncPolicy must be specified.");
        if (this.rotationPolicy == null) throw new IllegalStateException("RotationPolicy must be specified.");
        if (this.fsUrl == null) {
            throw new IllegalStateException("File system URL must be specified.");
        }

        this.collector = collector;
        this.fileNameFormat.prepare(conf, topologyContext);
        this.hdfsConfig = new Configuration();
        Map<String, Object> map = (Map<String, Object>)conf.get(this.configKey);
        if(map != null){
            for(String key : map.keySet()){
                this.hdfsConfig.set(key, String.valueOf(map.get(key)));
            }
        }

        try{
            HdfsSecurityUtil.login(conf, hdfsConfig);
            doPrepare(conf, topologyContext, collector);
            this.currentFile = createOutputFile();

        } catch (Exception e){
            throw new RuntimeException("Error preparing HdfsBolt: " + e.getMessage(), e);
        }

        if(this.rotationPolicy instanceof TimedRotationPolicy){
            long interval = ((TimedRotationPolicy)this.rotationPolicy).getInterval();
            this.rotationTimer = new Timer(true);
            TimerTask task = new TimerTask() {
                @Override
                public void run() {
                    try {
                        rotateOutputFile();
                    } catch(IOException e){
                        LOG.warn("IOException during scheduled file rotation.", e);
                    }
                }
            };
            this.rotationTimer.scheduleAtFixedRate(task, interval, interval);
        }
    }
890703
public void initTopology(String topologyName, int delay) {
        if (this.topology != null) {
            // TODO: possible refactor this code later
            System.out.println("Topology has been initialized before!");
            return;
        }
        this.topology = componentFactory.createTopology(topologyName);
    }
732491
@Override
  public void sourcePrepare(FlowProcess<? extends Configuration> flowProcess, SourceCall<Corc, RecordReader> sourceCall)
      throws IOException {
    sourceCall.setContext((Corc) sourceCall.getInput().createValue());
  }
468898
private void setup(final Handler<AsyncResult<Void>> doneHandler) {
    // Retrieve the component context from the coordinator (the current cluster).
    // If the context has changed due to a network configuration change, the
    // internal context and input/output connections will be automatically updated.
    log.debug(String.format("%s - Starting cluster coordination", DefaultComponent.this));
    coordinator.start(new Handler<AsyncResult<Void>>() {
      @Override
      public void handle(AsyncResult<Void> result) {
        if (result.failed()) {
          new DefaultFutureResult<Void>(result.cause()).setHandler(doneHandler);
        } else {
          // We have to make sure the input and output collectors are started
          // simultaneously in order to support circular connections. If both
          // input and output aren't started at the same time then circular
          // connections will never open.
          final CountingCompletionHandler<Void> ioHandler = new CountingCompletionHandler<Void>(2);
          ioHandler.setHandler(new Handler<AsyncResult<Void>>() {
            @Override
            public void handle(AsyncResult<Void> result) {
              if (result.failed()) {
                new DefaultFutureResult<Void>(result.cause()).setHandler(doneHandler);
              } else {
                // Tell the coordinator we're ready for the network to start.
                coordinator.resume();
              }
            }
          });

          output.open(new Handler<AsyncResult<Void>>() {
            @Override
            public void handle(AsyncResult<Void> result) {
              if (result.failed()) {
                log.error(String.format("%s - Failed to open component outputs", DefaultComponent.this), result.cause());
                ioHandler.fail(result.cause());
              } else {
                ioHandler.succeed();
              }
            }
          });

          input.open(new Handler<AsyncResult<Void>>() {
            @Override
            public void handle(AsyncResult<Void> result) {
              if (result.failed()) {
                log.error(String.format("%s - Failed to open component inputs", DefaultComponent.this), result.cause());
                ioHandler.fail(result.cause());
              } else {
                ioHandler.succeed();
              }
            }
          });
        }
      }
    });

    // The resume handler will be called by the coordinator once the
    // network's manager has indicated that all the components in the
    // network have finished setting up their connections.
    coordinator.resumeHandler(new Handler<Void>() {
      @Override
      @SuppressWarnings("unchecked")
      public void handle(Void _) {
        if (!started) {
          started = true;
          log.debug(String.format("%s - Started", DefaultComponent.this, context.component().name(), context.number()));
          List<ComponentHook> hooks = context.component().hooks();
          for (ComponentHook hook : hooks) {
            hook.handleStart(DefaultComponent.this);
          }
          new DefaultFutureResult<Void>((Void) null).setHandler(doneHandler);
        }
      }
    });
  }
1183533
private void updateNetwork(final NetworkContext context, final Handler<AsyncResult<Void>> doneHandler) {
    final CountingCompletionHandler<Void> complete = new CountingCompletionHandler<Void>(context.components().size());
    complete.setHandler(new Handler<AsyncResult<Void>>() {
      @Override
      public void handle(AsyncResult<Void> result) {
        if (result.failed()) {
          new DefaultFutureResult<Void>(result.cause()).setHandler(doneHandler);
        } else {
          new DefaultFutureResult<Void>((Void) null).setHandler(doneHandler);
        }
      }
    });
    updateComponents(context.components(), complete);
  }
485007
public <T> Collector<T> create(long id, Collection<Address> backupOwners, int topologyId) {
      if (backupOwners.isEmpty()) {
         return new PrimaryOwnerOnlyCollector<>();
      }
      SingleKeyCollector<T> collector = new SingleKeyCollector<>(id, backupOwners, topologyId);
      BaseAckTarget prev = collectorMap.put(id, collector);
      //is it possible the have a previous collector when the topology changes after the first collector is created
      //in that case, the previous collector must have a lower topology id
      assert prev == null || prev.topologyId < topologyId : format("replaced old collector '%s' by '%s'", prev, collector);
      if (trace) {
         log.tracef("Created new collector for %s. BackupOwners=%s", id, backupOwners);
      }
      return collector;
   }
1339391
private static void doTopologyLevelTranslation(Config heronConfig) {
    if (heronConfig.containsKey(org.apache.storm.Config.TOPOLOGY_ACKER_EXECUTORS)) {
      Integer nAckers =
          Utils.getInt(heronConfig.get(org.apache.storm.Config.TOPOLOGY_ACKER_EXECUTORS));
      if (nAckers > 0) {
        org.apache.heron.api.Config.setTopologyReliabilityMode(heronConfig,
                 org.apache.heron.api.Config.TopologyReliabilityMode.ATLEAST_ONCE);
      } else {
        org.apache.heron.api.Config.setTopologyReliabilityMode(heronConfig,
                 org.apache.heron.api.Config.TopologyReliabilityMode.ATMOST_ONCE);
      }
    } else {
      org.apache.heron.api.Config.setTopologyReliabilityMode(heronConfig,
               org.apache.heron.api.Config.TopologyReliabilityMode.ATMOST_ONCE);
    }
  }
479010
void open(TopologyContext context, SpoutOutputCollector collector,
              int checkpointInterval, KeyValueState<String, CheckPointState> checkpointState) {
        this.context = context;
        this.collector = collector;
        this.checkpointInterval = checkpointInterval;
        this.sleepInterval = checkpointInterval / 10;
        this.checkpointState = checkpointState;
        this.curTxState = checkpointState.get(TX_STATE_KEY);
        lastCheckpointTs = 0;
        recoveryStepInProgress = false;
        checkpointStepInProgress = false;
        recovering = true;
    }
594981
@Override
	public void init(Config config, TaskContext context) throws Exception {
		String yarnConfHome = config.get(SamzaConfigFactory.YARN_CONF_HOME_KEY);
		if (yarnConfHome != null && yarnConfHome.length() > 0) // if the property is set , otherwise, assume we are running in
            												// local mode and ignore this
			SystemsUtils.setHadoopConfigHome(yarnConfHome);
		
		String filename = config.get(SamzaConfigFactory.FILE_KEY);
		String filesystem = config.get(SamzaConfigFactory.FILESYSTEM_KEY);
		
		this.setName(config.get(SamzaConfigFactory.JOB_NAME_KEY));
		SerializationProxy wrapper = (SerializationProxy) SystemsUtils.deserializeObjectFromFileAndKey(filesystem, filename, this.getName());
		this.setOutputStream(wrapper.outputStream);
		SamzaStream output = (SamzaStream)this.getOutputStream();
		if (output != null) // if output stream exists, set it up
			output.onCreate();
	}
879127
@SuppressWarnings("unchecked")
  public static void registerTopologyTimerEvents(Map<String, Object> conf,
                                                 String name, Duration interval,
                                                 Runnable task) {
    if (interval.isZero() || interval.isNegative()) {
      throw new IllegalArgumentException("Timer duration needs to be positive");
    }
    if (!conf.containsKey(Config.TOPOLOGY_TIMER_EVENTS)) {
      conf.put(Config.TOPOLOGY_TIMER_EVENTS, new HashMap<String, Pair<Duration, Runnable>>());
    }

    Map<String, Pair<Duration, Runnable>> timers
        = (Map<String, Pair<Duration, Runnable>>) conf.get(Config.TOPOLOGY_TIMER_EVENTS);

    if (timers.containsKey(name)) {
      throw new IllegalArgumentException("Timer with name " + name + " already exists");
    }
    timers.put(name, Pair.of(interval, task));
  }
584502
public void submitTopology() throws TopologySubmissionException {
    // build primary runtime config first
    Config primaryRuntime = Config.newBuilder()
          .putAll(LauncherUtils.getInstance().createPrimaryRuntime(topology)).build();
    // call launcher directly here if in dry-run mode
    if (Context.dryRun(config)) {
      callLauncherRunner(primaryRuntime);
      return;
    }
    // 1. Do prepare work
    // create an instance of state manager
    String statemgrClass = Context.stateManagerClass(config);
    IStateManager statemgr;

    // Create an instance of the launcher class
    String launcherClass = Context.launcherClass(config);
    ILauncher launcher;

    // create an instance of the uploader class
    String uploaderClass = Context.uploaderClass(config);
    IUploader uploader;

    // create an instance of state manager
    try {
      statemgr = ReflectionUtils.newInstance(statemgrClass);
    } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
      throw new TopologySubmissionException(
          String.format("Failed to instantiate state manager class '%s'", statemgrClass), e);
    }

    // create an instance of launcher
    try {
      launcher = ReflectionUtils.newInstance(launcherClass);
    } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
      throw new LauncherException(
          String.format("Failed to instantiate launcher class '%s'", launcherClass), e);
    }

    // create an instance of uploader
    try {
      uploader = ReflectionUtils.newInstance(uploaderClass);
    } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
      throw new UploaderException(
          String.format("Failed to instantiate uploader class '%s'", uploaderClass), e);
    }

    // Put it in a try block so that we can always clean resources
    try {
      // initialize the state manager
      statemgr.initialize(config);

      // initialize the uploader
      uploader.initialize(config);

      // TODO(mfu): timeout should read from config
      SchedulerStateManagerAdaptor adaptor = new SchedulerStateManagerAdaptor(statemgr, 5000);

      // Check if topology is already running
      validateSubmit(adaptor, topology.getName());

      LOG.log(Level.FINE, "Topology {0} to be submitted", topology.getName());

      Config runtimeWithoutPackageURI = Config.newBuilder()
          .putAll(primaryRuntime)
          .putAll(LauncherUtils.getInstance().createAdaptorRuntime(adaptor))
          .put(Key.LAUNCHER_CLASS_INSTANCE, launcher)
          .build();

      PackingPlan packingPlan = LauncherUtils.getInstance()
          .createPackingPlan(config, runtimeWithoutPackageURI);

      // The packing plan might call for a number of containers different than the config
      // settings. If that's the case we need to modify the configs to match.
      runtimeWithoutPackageURI =
          updateNumContainersIfNeeded(runtimeWithoutPackageURI, topology, packingPlan);

      // If the packing plan is valid we will upload necessary packages
      URI packageURI = uploadPackage(uploader);

      // Update the runtime config with the packageURI
      Config runtimeAll = Config.newBuilder()
          .putAll(runtimeWithoutPackageURI)
          .put(Key.TOPOLOGY_PACKAGE_URI, packageURI)
          .build();

      callLauncherRunner(runtimeAll);

    } catch (LauncherException | PackingException e) {
      // we undo uploading of topology package only if launcher fails to
      // launch topology, which will throw LauncherException or PackingException
      uploader.undo();
      throw e;
    } finally {
      SysUtils.closeIgnoringExceptions(uploader);
      SysUtils.closeIgnoringExceptions(launcher);
      SysUtils.closeIgnoringExceptions(statemgr);
    }
  }
339303
public static void downloadDistributeStormCode(Map conf, String topologyId, String masterCodeDir)
            throws IOException, TException {
        String tmpToot = null;
        try {
            // STORM_LOCAL_DIR/supervisor/tmp/(UUID)
            tmpToot = StormConfig.supervisorTmpDir(conf) + File.separator + UUID.randomUUID().toString();

            // STORM_LOCAL_DIR/supervisor/stormdist/topologyId
            String stormRoot = StormConfig.supervisor_stormdist_root(conf, topologyId);

//        JStormServerUtils.downloadCodeFromMaster(conf, tmproot, masterCodeDir, topologyId, true);
            JStormServerUtils.downloadCodeFromBlobStore(conf, tmpToot, topologyId);

            // tmproot/stormjar.jar
            String localFileJarTmp = StormConfig.stormjar_path(tmpToot);

            // extract dir from jar
            JStormUtils.extractDirFromJar(localFileJarTmp, StormConfig.RESOURCES_SUBDIR, tmpToot);

            File srcDir = new File(tmpToot);
            File destDir = new File(stormRoot);
            try {
                FileUtils.moveDirectory(srcDir, destDir);
            } catch (FileExistsException e) {
                FileUtils.copyDirectory(srcDir, destDir);
                FileUtils.deleteQuietly(srcDir);
            }
        } finally {
            if (tmpToot != null) {
                File srcDir = new File(tmpToot);
                FileUtils.deleteQuietly(srcDir);
            }
        }
    }
1045851
@SuppressWarnings("rawtypes")
    public void initialize(Map conf, TopologyContext context, String function)
    {
        this.context = context;
        this.stormConf = conf;
        this.function = function;
        this.idsMap = new RotatingMap<TransactionAttempt, DrpcRequestInfo>(ROTATE_SIZE);
        this.rotateTime = TimeUnit.SECONDS.toMillis(((Number) conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS)).intValue());
        this.lastRotate = getCurrentTime();
    }
898105
private void configure(Member.Type type, CompletableFuture<Void> future) {
    // Set a timer to retry the attempt to leave the cluster.
    configureTimeout = cluster.getContext().getThreadContext().schedule(cluster.getContext().getElectionTimeout(), () -> {
      configure(type, future);
    });

    // Attempt to leave the cluster by submitting a LeaveRequest directly to the server state.
    // Non-leader states should forward the request to the leader if there is one. Leader states
    // will log, replicate, and commit the reconfiguration.
    cluster.getContext().getServerState().reconfigure(ReconfigureRequest.builder()
      .withIndex(cluster.getConfiguration().index())
      .withTerm(cluster.getConfiguration().term())
      .withMember(new ServerMember(type, serverAddress(), clientAddress(), updated))
      .build()).whenComplete((response, error) -> {
      if (error == null) {
        if (response.status() == Response.Status.OK) {
          cancelConfigureTimer();
          cluster.configure(new Configuration(response.index(), response.term(), response.timestamp(), response.members()));
          future.complete(null);
        } else if (response.error() == null || response.error() == CopycatError.Type.NO_LEADER_ERROR) {
          cancelConfigureTimer();
          configureTimeout = cluster.getContext().getThreadContext().schedule(cluster.getContext().getElectionTimeout().multipliedBy(2), () -> configure(type, future));
        } else {
          cancelConfigureTimer();
          future.completeExceptionally(response.error().createException());
        }
      }
    });
  }
344486
private void processInitializedConduits(Source source) {
        String sourceId = CollectorManagerUtils.getSourceId(source);
        String sourceName = source.getSourceName();
        /*
         * Check that the current source initialized is Log/Message and Trace.
         * These are the unique special sources along with their respective Conduits/BufferManagers
         * that were started before OSGI and thusly were not configured via 'the osgi' way that CollectorManager
         * had in place before the JSON Logging work.
         */
        if (sourceName.equals(CollectorConstants.MESSAGES_SOURCE) || sourceName.equals(CollectorConstants.TRACE_SOURCE)) {
            //Make sure we have a bundleContext, we need this to play with osgi (i.e. register/listen to services)
            if (bundleContext == null) {
                retrieveBundleContext();
            }
            /*
             * We should really check if the source and conduit/bufferManager is already set into the
             * SourceManager and BufferManager maps. If so, no need to continue the rest of this method.
             */
            if (sourceMgrs.containsKey(sourceId) && bufferManagerMap.containsKey(sourceId)) {
                return;
            }

            ServiceReference<BufferManager>[] servRefs;
            try {
                servRefs = (ServiceReference<BufferManager>[]) bundleContext.getServiceReferences(BufferManager.class.getName(), null);
                if (servRefs != null) {
                    for (ServiceReference<BufferManager> servRef : servRefs) {
                        /*
                         * Ensure that the Conduit/BufferManager retrieved is the right one for the source
                         * by checking that the 'source' that the conduit/bufferManager was assinged to
                         * was for the Source that we are registering.
                         */
                        if (sourceName.equals(servRef.getProperty("source"))) {
                            /*
                             * Retrieve the actual Conduit/BufferManager fro mthe service Reference
                             * and put it into the bufferManagerMap.
                             */
                            Object object = bundleContext.getService(servRef);
                            BufferManager conduit = (BufferManager) object;
                            bufferManagerMap.put(sourceId, conduit);
                        }
                    }
                }

            } catch (InvalidSyntaxException e) {
                e.printStackTrace();
            }
        }
    }
1284854
void onNextJVM(final AllocatedEvaluator allocatedEvaluator) {
    try {
      final Configuration contextConfiguration = ContextConfiguration.CONF
          .set(ContextConfiguration.IDENTIFIER, "HelloREEFContext")
          .build();

      final Configuration taskConfiguration = TaskConfiguration.CONF
          .set(TaskConfiguration.IDENTIFIER, "HelloREEFTask")
          .set(TaskConfiguration.TASK, HelloTask.class)
          .build();

      allocatedEvaluator.submitContextAndTask(contextConfiguration, taskConfiguration);
    } catch (final BindException ex) {
      final String message = "Unable to setup Task or Context configuration.";
      LOG.log(Level.SEVERE, message, ex);
      throw new RuntimeException(message, ex);
    }
  }
551785
private void handleCreate(final NetworkContext context) {
    log.debug(String.format("%s - Network configuration created in cluster: %s", NetworkManager.this, cluster.address()));
    log.debug(String.format("%s - Scheduling network deployment task", NetworkManager.this));
    tasks.runTask(new Handler<Task>() {
      @Override
      public void handle(final Task task) {
        currentContext = context;

        // Any time the network is being reconfigured, unready the network.
        // This will cause components to pause during the reconfiguration.
        unready(new Handler<AsyncResult<Void>>() {
          @Override
          public void handle(AsyncResult<Void> result) {
            if (result.failed()) {
              log.error(result.cause());
              task.complete();
            } else {
              deployNetwork(context, new Handler<AsyncResult<NetworkContext>>() {
                @Override
                public void handle(AsyncResult<NetworkContext> result) {
                  task.complete();
                  if (result.failed()) {
                    log.error(result.cause());
                  } else {
                    log.info(String.format("%s - Successfully deployed network", NetworkManager.this));
                  }
                }
              });
            }
          }
        });
      }
    });
  }