text:
package kafka.manager

import java.util.Properties
import java.util.concurrent.{LinkedBlockingQueue, ThreadPoolExecutor, TimeUnit}
import org.json4s.jackson.JsonMethods.parse

import akka.actor.{ActorPath, ActorSystem, Cancellable, Props}
import akka.util.Timeout
import com.typesafe.config.{Config, ConfigFactory}
import grizzled.slf4j.Logging
import kafka.manager.actor.{KafkaManagerActor, KafkaManagerActorConfig}
import kafka.manager.base.LongRunningPoolConfig
import kafka.manager.model._
import ActorModel._
import kafka.manager.actor.cluster.KafkaManagedOffsetCacheConfig
import kafka.manager.utils.UtilException
import kafka.manager.utils.zero81.ReassignPartitionErrors.ReplicationOutOfSync
import kafka.manager.utils.zero81.{ForceOnReplicationOutOfSync, ForceReassignmentCommand, ReassignPartitionErrors}

import scala.concurrent.{Await, ExecutionContext, Future}
import scala.concurrent.duration._
import scala.reflect.ClassTag
import scala.util.{Failure, Success, Try}

/**
 * @author hiral
 */
case class TopicListExtended(list: IndexedSeq[(String, Option[TopicIdentity])],
                             topicToConsumerMap: Map[String, Iterable[(String, ConsumerType)]],
                             deleteSet: Set[String],
                             underReassignments: IndexedSeq[String],
                             clusterContext: ClusterContext)
case class BrokerListExtended(list: IndexedSeq[BrokerIdentity],
                              metrics: Map[Int,BrokerMetrics],
                              combinedMetric: Option[BrokerMetrics],
                              clusterContext: ClusterContext)
case class ConsumerListExtended(list: IndexedSeq[((String, ConsumerType), Option[ConsumerIdentity])], clusterContext: ClusterContext)
case class LogkafkaListExtended(list: IndexedSeq[(String, Option[LogkafkaIdentity])], deleteSet: Set[String])

case class ApiError(msg: String, recoverByForceOperation: Boolean = false)
object ApiError extends Logging {

  implicit def fromThrowable(t: Throwable) : ApiError = {
    error(s"error : ${t.getMessage}", t)
    ApiError(t.getMessage)
  }

  implicit def from(actorError: ActorErrorResponse): ApiError = {
    actorError.throwableOption.foreach { t =>
      error(s"Actor error : ${actorError.msg}", t)
    }
    ApiError(actorError.msg)
  }
}

import akka.pattern._
import scalaz.{-\/, \/, \/-}
class KafkaManager(akkaConfig: Config) extends Logging {

  def getPrefixedKey(key: String): String = if (akkaConfig.hasPathOrNull(s"cmak.$key")) s"cmak.$key" else s"kafka-manager.$key"

  val ConsumerPropertiesFile = getPrefixedKey("consumer.properties.file")
  val BaseZkPath = getPrefixedKey("base-zk-path")
  val PinnedDispatchName = getPrefixedKey("pinned-dispatcher-name")
  val ZkHosts = getPrefixedKey("zkhosts")
  val BrokerViewUpdateSeconds = getPrefixedKey("broker-view-update-seconds")
  val KafkaManagerUpdateSeconds = getPrefixedKey("kafka-manager-update-seconds")
  val DeleteClusterUpdateSeconds = getPrefixedKey("delete-cluster-update-seconds")
  val DeletionBatchSize = getPrefixedKey("deletion-batch-size")
  val MaxQueueSize = getPrefixedKey("max-queue-size")
  val ThreadPoolSize = getPrefixedKey("thread-pool-size")
  val MutexTimeoutMillis = getPrefixedKey("mutex-timeout-millis")
  val StartDelayMillis = getPrefixedKey("start-delay-millis")
  val ApiTimeoutMillis = getPrefixedKey("api-timeout-millis")
  val ClusterActorsAskTimeoutMillis = getPrefixedKey("cluster-actors-ask-timeout-millis")
  val PartitionOffsetCacheTimeoutSecs = getPrefixedKey("partition-offset-cache-timeout-secs")
  val SimpleConsumerSocketTimeoutMillis = getPrefixedKey("simple-consumer-socket-timeout-millis")
  val BrokerViewThreadPoolSize = getPrefixedKey("broker-view-thread-pool-size")
  val BrokerViewMaxQueueSize = getPrefixedKey("broker-view-max-queue-size")
  val OffsetCacheThreadPoolSize = getPrefixedKey("offset-cache-thread-pool-size")
  val OffsetCacheMaxQueueSize = getPrefixedKey("offset-cache-max-queue-size")
  val KafkaAdminClientThreadPoolSize = getPrefixedKey("kafka-admin-client-thread-pool-size")
  val KafkaAdminClientMaxQueueSize = getPrefixedKey("kafka-admin-client-max-queue-size")
  val KafkaManagedOffsetMetadataCheckMillis = getPrefixedKey("kafka-managed-offset-metadata-check-millis")
  val KafkaManagedOffsetGroupCacheSize = getPrefixedKey("kafka-managed-offset-group-cache-size")
  val KafkaManagedOffsetGroupExpireDays = getPrefixedKey("kafka-managed-offset-group-expire-days")

  val DefaultConfig: Config = {
    val defaults: Map[String, _ <: AnyRef] = Map(
      BaseZkPath -> KafkaManagerActor.ZkRoot,
      PinnedDispatchName -> "pinned-dispatcher",
      BrokerViewUpdateSeconds -> "30",
      KafkaManagerUpdateSeconds -> "10",
      DeleteClusterUpdateSeconds -> "10",
      DeletionBatchSize -> "2",
      MaxQueueSize -> "100",
      ThreadPoolSize -> "2",
      MutexTimeoutMillis -> "4000",
      StartDelayMillis -> "1000",
      ApiTimeoutMillis -> "5000",
      ClusterActorsAskTimeoutMillis -> "2000",
      PartitionOffsetCacheTimeoutSecs -> "5",
      SimpleConsumerSocketTimeoutMillis -> "10000",
      BrokerViewThreadPoolSize -> Runtime.getRuntime.availableProcessors().toString,
      BrokerViewMaxQueueSize -> "1000",
      OffsetCacheThreadPoolSize -> Runtime.getRuntime.availableProcessors().toString,
      OffsetCacheMaxQueueSize -> "1000",
      KafkaAdminClientThreadPoolSize -> Runtime.getRuntime.availableProcessors().toString,
      KafkaAdminClientMaxQueueSize -> "1000",
      KafkaManagedOffsetMetadataCheckMillis -> KafkaManagedOffsetCacheConfig.defaultGroupMemberMetadataCheckMillis.toString,
      KafkaManagedOffsetGroupCacheSize -> KafkaManagedOffsetCacheConfig.defaultGroupTopicPartitionOffsetMaxSize.toString,
      KafkaManagedOffsetGroupExpireDays -> KafkaManagedOffsetCacheConfig.defaultGroupTopicPartitionOffsetExpireDays.toString
    )
    import scala.collection.JavaConverters._
    ConfigFactory.parseMap(defaults.asJava)
  }

  private[this] val system = ActorSystem("kafka-manager-system", akkaConfig)

  private[this] val configWithDefaults = akkaConfig.withFallback(DefaultConfig)
  val defaultTuning = ClusterTuning(
    brokerViewUpdatePeriodSeconds = Option(configWithDefaults.getInt(BrokerViewUpdateSeconds))
    , clusterManagerThreadPoolSize = Option(configWithDefaults.getInt(ThreadPoolSize))
    , clusterManagerThreadPoolQueueSize = Option(configWithDefaults.getInt(MaxQueueSize))
    , kafkaCommandThreadPoolSize = Option(configWithDefaults.getInt(ThreadPoolSize))
    , kafkaCommandThreadPoolQueueSize = Option(configWithDefaults.getInt(MaxQueueSize))
    , logkafkaCommandThreadPoolSize = Option(configWithDefaults.getInt(ThreadPoolSize))
    , logkafkaCommandThreadPoolQueueSize = Option(configWithDefaults.getInt(MaxQueueSize))
    , logkafkaUpdatePeriodSeconds = Option(configWithDefaults.getInt(BrokerViewUpdateSeconds))
    , partitionOffsetCacheTimeoutSecs = Option(configWithDefaults.getInt(PartitionOffsetCacheTimeoutSecs))
    , brokerViewThreadPoolSize = Option(configWithDefaults.getInt(BrokerViewThreadPoolSize))
    , brokerViewThreadPoolQueueSize = Option(configWithDefaults.getInt(BrokerViewMaxQueueSize))
    , offsetCacheThreadPoolSize = Option(configWithDefaults.getInt(OffsetCacheThreadPoolSize))
    , offsetCacheThreadPoolQueueSize = Option(configWithDefaults.getInt(OffsetCacheMaxQueueSize))
    , kafkaAdminClientThreadPoolSize = Option(configWithDefaults.getInt(KafkaAdminClientThreadPoolSize))
    , kafkaAdminClientThreadPoolQueueSize = Option(configWithDefaults.getInt(KafkaAdminClientMaxQueueSize))
    , kafkaManagedOffsetMetadataCheckMillis = Option(configWithDefaults.getInt(KafkaManagedOffsetMetadataCheckMillis))
    , kafkaManagedOffsetGroupCacheSize = Option(configWithDefaults.getInt(KafkaManagedOffsetGroupCacheSize))
    , kafkaManagedOffsetGroupExpireDays = Option(configWithDefaults.getInt(KafkaManagedOffsetGroupExpireDays))
  )
  private[this] val kafkaManagerConfig = {
    val curatorConfig = CuratorConfig(configWithDefaults.getString(ZkHosts))
    KafkaManagerActorConfig(
      curatorConfig = curatorConfig
      , baseZkPath = configWithDefaults.getString(BaseZkPath)
      , pinnedDispatcherName = configWithDefaults.getString(PinnedDispatchName)
      , startDelayMillis = configWithDefaults.getLong(StartDelayMillis)
      , threadPoolSize = configWithDefaults.getInt(ThreadPoolSize)
      , mutexTimeoutMillis = configWithDefaults.getInt(MutexTimeoutMillis)
      , maxQueueSize = configWithDefaults.getInt(MaxQueueSize)
      , kafkaManagerUpdatePeriod = FiniteDuration(configWithDefaults.getInt(KafkaManagerUpdateSeconds), SECONDS)
      , deleteClusterUpdatePeriod = FiniteDuration(configWithDefaults.getInt(DeleteClusterUpdateSeconds), SECONDS)
      , deletionBatchSize = configWithDefaults.getInt(DeletionBatchSize)
      , clusterActorsAskTimeoutMillis = configWithDefaults.getInt(ClusterActorsAskTimeoutMillis)
      , simpleConsumerSocketTimeoutMillis =  configWithDefaults.getInt(SimpleConsumerSocketTimeoutMillis)
      , defaultTuning = defaultTuning
      , consumerProperties = getConsumerPropertiesFromConfig(configWithDefaults)
    )
  }

  private[this] val props = Props(classOf[KafkaManagerActor], kafkaManagerConfig)

  private[this] val kafkaManagerActor: ActorPath = system.actorOf(props, "kafka-manager").path

  private[this] val apiExecutor = new ThreadPoolExecutor(
    kafkaManagerConfig.threadPoolSize,
    kafkaManagerConfig.threadPoolSize,
    0L,
    TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue[Runnable](kafkaManagerConfig.maxQueueSize)
  )

  private[this] val apiExecutionContext = ExecutionContext.fromExecutor(apiExecutor)

  private[this] implicit val apiTimeout: Timeout = FiniteDuration(
    configWithDefaults.getInt(ApiTimeoutMillis),
    MILLISECONDS
  )

  private[this] def getConsumerPropertiesFromConfig(config: Config) : Option[Properties] = {
    if(config.hasPath(ConsumerPropertiesFile)) {
      val filePath = config.getString(ConsumerPropertiesFile)
      val file = new java.io.File(filePath)
      if(file.isFile & file.canRead) {
        val props = new Properties()
        props.load(new java.io.FileInputStream(file))
        return Option(props)
      } else {
        warn(s"Failed to find consumer properties file or file is not readable : $file")
      }
    }
    None
  }

  private[this] def tryWithKafkaManagerActor[Input, Output, FOutput](msg: Input)
    (fn: Output => FOutput)
    (implicit tag: ClassTag[Output]): Future[ApiError \/ FOutput] =
  {
    implicit val ec = apiExecutionContext
    system.actorSelection(kafkaManagerActor).ask(msg).map {
      case err: ActorErrorResponse =>
        error(s"Failed on input : $msg")
        -\/(ApiError.from(err))
      case o: Output =>
        Try {
          fn(o)
        } match {
          case Failure(t) =>
            error(s"Failed on input : $msg")
            -\/(ApiError.fromThrowable(t))
          case Success(foutput) => \/-(foutput)
        }
    }.recover { case t: Throwable =>
      error(s"Failed on input : $msg", t)
      -\/(ApiError.fromThrowable(t))
    }
  }

  private[this] def withKafkaManagerActor[Input, Output, FOutput](msg: Input)
    (fn: Output => Future[ApiError \/ FOutput])
    (implicit tag: ClassTag[Output]): Future[ApiError \/ FOutput] =
  {
    implicit val ec = apiExecutionContext
    system.actorSelection(kafkaManagerActor).ask(msg).flatMap {
      case err: ActorErrorResponse => Future.successful(-\/(ApiError.from(err)))
      case o: Output =>
        fn(o)
    }.recover { case t: Throwable =>
      -\/(ApiError.fromThrowable(t))
    }
  }

  private[this] def toDisjunction[T](t: Try[T]): ApiError \/ T = {
    t match {
      case Failure(th) =>
        -\/(th)
      case Success(tInst) =>
        \/-(tInst)
    }
  }

  def shutdown(): Unit = {
    implicit val ec = apiExecutionContext
    system.actorSelection(kafkaManagerActor).tell(KMShutdown, system.deadLetters)
    Try(Await.ready(system.terminate(), Duration(30, TimeUnit.SECONDS)))
    apiExecutor.shutdown()
  }

  //--------------------Commands--------------------------
  def addCluster(clusterName: String,
                 version: String,
                 zkHosts: String,
                 jmxEnabled: Boolean,
                 jmxUser: Option[String],
                 jmxPass: Option[String],
                 jmxSsl: Boolean,
                 pollConsumers: Boolean,
                 filterConsumers: Boolean,
                 tuning: Option[ClusterTuning],
                 securityProtocol: String,
                 saslMechanism: Option[String],
                 jaasConfig: Option[String],
                 logkafkaEnabled: Boolean = false,
                 activeOffsetCacheEnabled: Boolean = false,
                 displaySizeEnabled: Boolean = false): Future[ApiError \/ Unit] =
  {
    val cc = ClusterConfig(
      clusterName,
      version,
      zkHosts,
      tuning = tuning,
      securityProtocol = securityProtocol,
      saslMechanism = saslMechanism,
      jaasConfig = jaasConfig,
      jmxEnabled = jmxEnabled,
      jmxUser = jmxUser,
      jmxPass = jmxPass,
      jmxSsl = jmxSsl,
      pollConsumers = pollConsumers,
      filterConsumers = filterConsumers,
      logkafkaEnabled = logkafkaEnabled,
      activeOffsetCacheEnabled = activeOffsetCacheEnabled,
      displaySizeEnabled = displaySizeEnabled)
    tryWithKafkaManagerActor(KMAddCluster(cc)) { result: KMCommandResult =>
      result.result.get
    }
  }

  def updateCluster(clusterName: String,
                    version: String,
                    zkHosts: String,
                    jmxEnabled: Boolean,
                    jmxUser: Option[String],
                    jmxPass: Option[String],
                    jmxSsl: Boolean,
                    pollConsumers: Boolean,
                    filterConsumers: Boolean,
                    tuning: Option[ClusterTuning],
                    securityProtocol: String,
                    saslMechanism: Option[String],
                    jaasConfig: Option[String],
                    logkafkaEnabled: Boolean = false,
                    activeOffsetCacheEnabled: Boolean = false,
                    displaySizeEnabled: Boolean = false): Future[ApiError \/ Unit] =
  {
    val cc = ClusterConfig(
      clusterName,
      version,
      zkHosts,
      tuning = tuning,
      securityProtocol = securityProtocol,
      saslMechanism = saslMechanism,
      jaasConfig = jaasConfig,
      jmxEnabled = jmxEnabled,
      jmxUser = jmxUser,
      jmxPass = jmxPass,
      jmxSsl = jmxSsl,
      pollConsumers = pollConsumers,
      filterConsumers = filterConsumers,
      logkafkaEnabled = logkafkaEnabled,
      activeOffsetCacheEnabled = activeOffsetCacheEnabled,
      displaySizeEnabled = displaySizeEnabled)
    tryWithKafkaManagerActor(KMUpdateCluster(cc)) { result: KMCommandResult =>
      result.result.get
    }
  }

  def disableCluster(clusterName: String): Future[ApiError \/ Unit] = {
    tryWithKafkaManagerActor(KMDisableCluster(clusterName)) { result: KMCommandResult =>
      result.result.get
    }
  }

  def enableCluster(clusterName: String): Future[ApiError \/ Unit] = {
    tryWithKafkaManagerActor(KMEnableCluster(clusterName)) { result: KMCommandResult =>
      result.result.get
    }
  }

  def deleteCluster(clusterName: String): Future[ApiError \/ Unit] = {
    tryWithKafkaManagerActor(KMDeleteCluster(clusterName)) { result: KMCommandResult =>
      result.result.get
    }
  }

  def runPreferredLeaderElection(clusterName: String, topics: Set[String]): Future[ApiError \/ ClusterContext] = {
    implicit val ec = apiExecutionContext
    withKafkaManagerActor(
      KMClusterCommandRequest(
        clusterName,
        CMRunPreferredLeaderElection(topics)
      )
    ) { result: Future[CMCommandResult] =>
      result.map(cmr => toDisjunction(cmr.result))
    }
  }

  private def runPreferredLeaderElectionWithAllTopics(clusterName: String) = {
    implicit val ec = apiExecutionContext

    getTopicList(clusterName).flatMap { errorOrTopicList =>
      errorOrTopicList.fold({ e =>
        Future.successful(-\/(e))
      }, { topicList =>
        runPreferredLeaderElection(clusterName, topicList.list.toSet)
      })
    }
  }

  private def updateSchedulePreferredLeaderElection(clusterName: String): Unit = {
    system.actorSelection(kafkaManagerActor).ask(KMClusterCommandRequest(
      clusterName,
      CMSchedulePreferredLeaderElection(
        pleCancellable map { case (key, value) => (key, value._2) }
      )
    ))
  }

  def schedulePreferredLeaderElection(clusterName: String, topics: Set[String], timeIntervalMinutes: Int): Future[String] = {
    implicit val ec = apiExecutionContext

    pleCancellable += (clusterName ->
      (
        Some(
          system.scheduler.schedule(0 seconds, Duration(timeIntervalMinutes, TimeUnit.MINUTES)) {
            runPreferredLeaderElectionWithAllTopics(clusterName)
          }
        ),
        timeIntervalMinutes
      )
    )
    updateSchedulePreferredLeaderElection(clusterName)

    Future("Scheduler started")
  }

  def cancelPreferredLeaderElection(clusterName: String): Future[String] = {
    implicit val ec = apiExecutionContext

    pleCancellable(clusterName)._1.map(_.cancel())
    pleCancellable -= clusterName
    updateSchedulePreferredLeaderElection(clusterName)
    Future("Scheduler stopped")
  }

  def manualPartitionAssignments(clusterName: String,
                                 assignments: List[(String, List[(Int, List[Int])])]) = {
    implicit val ec = apiExecutionContext
    val results = tryWithKafkaManagerActor(
      KMClusterCommandRequest (
        clusterName,
        CMManualPartitionAssignments(assignments)
      )
    ) { result: CMCommandResults =>
      val errors = result.result.collect { case Failure(t) => ApiError(t.getMessage)}
      if (errors.isEmpty)
        \/-({})
      else
        -\/(errors)
    }

    results.map {
      case -\/(e) => -\/(IndexedSeq(e))
      case \/-(lst) => lst
    }
  }

  def generatePartitionAssignments(
                                    clusterName: String,
                                    topics: Set[String],
                                    brokers: Set[Int],
                                    replicationFactor: Option[Int] = None
                                    ): Future[IndexedSeq[ApiError] \/ Unit] =
  {
    val results = tryWithKafkaManagerActor(
      KMClusterCommandRequest(
        clusterName,
        CMGeneratePartitionAssignments(topics, brokers, replicationFactor)
      )
    ) { result: CMCommandResults =>
      val errors = result.result.collect { case Failure(t) => ApiError(t.getMessage)}
      if (errors.isEmpty)
        \/-({})
      else
        -\/(errors)
    }
    implicit val ec = apiExecutionContext
    results.map {
      case -\/(e) => -\/(IndexedSeq(e))
      case \/-(lst) => lst
    }
  }

  def runReassignPartitions(clusterName: String, topics: Set[String], force: Boolean = false): Future[IndexedSeq[ApiError] \/ Unit] = {
    implicit val ec = apiExecutionContext
    val forceSet: Set[ForceReassignmentCommand] = {
      if(force) {
        Set(ForceOnReplicationOutOfSync)
      } else Set.empty
    }
    val results = tryWithKafkaManagerActor(KMClusterCommandRequest(clusterName, CMRunReassignPartition(topics, forceSet))) {
      resultFuture: Future[CMCommandResults] =>
        resultFuture map { result =>
          val errors = result.result.collect {
            case Failure(t) =>
              t match {
                case UtilException(e) if e.isInstanceOf[ReplicationOutOfSync] =>
                  ApiError(t.getMessage, recoverByForceOperation = true)
                case _ =>
                  ApiError(t.getMessage)
              }
          }
          if (errors.isEmpty)
            \/-({})
          else
            -\/(errors)
        }
    }
    results.flatMap {
      case \/-(lst) => lst
      case -\/(e) => Future.successful(-\/(IndexedSeq(e)))
    }
  }

  def createTopic(
                   clusterName: String,
                   topic: String,
                   partitions: Int,
                   replication: Int,
                   config: Properties = new Properties
                   ): Future[ApiError \/ ClusterContext] =
  {
    implicit val ec = apiExecutionContext
    withKafkaManagerActor(KMClusterCommandRequest(clusterName, CMCreateTopic(topic, partitions, replication, config))) {
      result: Future[CMCommandResult] =>
        result.map(cmr => toDisjunction(cmr.result))
    }
  }

  def addTopicPartitions(
                          clusterName: String,
                          topic: String,
                          brokers: Seq[Int],
                          partitions: Int,
                          readVersion: Int
                          ): Future[ApiError \/ ClusterContext] =
  {
    implicit val ec = apiExecutionContext
    getTopicIdentity(clusterName, topic).flatMap { topicIdentityOrError =>
      topicIdentityOrError.fold(
      e => Future.successful(-\/(e)), { ti =>
        val partitionReplicaList: Map[Int, Seq[Int]] = ti.partitionsIdentity.mapValues(_.replicas)
        withKafkaManagerActor(
          KMClusterCommandRequest(
            clusterName,
            CMAddTopicPartitions(topic, brokers, partitions, partitionReplicaList, readVersion)
          )
        ) {
          result: Future[CMCommandResult] =>
            result.map(cmr => toDisjunction(cmr.result))
        }
      }
      )
    }
  }

  def addMultipleTopicsPartitions(
                              clusterName: String,
                              topics: Seq[String],
                              brokers: Set[Int],
                              partitions: Int,
                              readVersions: Map[String, Int]
                              ): Future[ApiError \/ ClusterContext] =
  {
    implicit val ec = apiExecutionContext
    getTopicListExtended(clusterName).flatMap { tleOrError =>
      tleOrError.fold(
      e => Future.successful(-\/(e)), { tle =>
        // add partitions to only topics with topic identity
        val topicsAndReplicas = topicListSortedByNumPartitions(tle).filter(t => topics.contains(t._1) && t._2.nonEmpty).map{ case (t,i) => (t, i.get.partitionsIdentity.mapValues(_.replicas)) }
        withKafkaManagerActor(
          KMClusterCommandRequest(
            clusterName,
            CMAddMultipleTopicsPartitions(topicsAndReplicas, brokers, partitions, readVersions)
          )
        ) {
          result: Future[CMCommandResult] =>
            result.map(cmr => toDisjunction(cmr.result))
        }
      }
      )
    }
  }

  def updateTopicConfig(
                         clusterName: String,
                         topic: String,
                         config: Properties,
                         readVersion: Int
                         ): Future[ApiError \/ ClusterContext] =
  {
    implicit val ec = apiExecutionContext
    withKafkaManagerActor(
      KMClusterCommandRequest(
        clusterName,
        CMUpdateTopicConfig(topic, config, readVersion)
      )
    ) {
      result: Future[CMCommandResult] =>
        result.map(cmr => toDisjunction(cmr.result))
    }
  }

  def deleteTopic(
                   clusterName: String,
                   topic: String
                   ): Future[ApiError \/ ClusterContext] =
  {
    implicit val ec = apiExecutionContext
    withKafkaManagerActor(KMClusterCommandRequest(clusterName, CMDeleteTopic(topic))) {
      result: Future[CMCommandResult] =>
        result.map(cmr => toDisjunction(cmr.result))
    }
  }

  def createLogkafka(
                   clusterName: String,
                   logkafka_id: String,
                   log_path: String,
                   config: Properties = new Properties
                   ): Future[ApiError \/ ClusterContext] =
  {
    implicit val ec = apiExecutionContext
    withKafkaManagerActor(KMClusterCommandRequest(clusterName, CMCreateLogkafka(logkafka_id, log_path, config))) {
      result: Future[CMCommandResult] =>
        result.map(cmr => toDisjunction(cmr.result))
    }
  }

  def updateLogkafkaConfig(
                         clusterName: String,
                         logkafka_id: String,
                         log_path: String,
                         config: Properties,
                         checkConfig: Boolean = true
                         ): Future[ApiError \/ ClusterContext] =
  {
    implicit val ec = apiExecutionContext
    withKafkaManagerActor(
      KMClusterCommandRequest(
        clusterName,
        CMUpdateLogkafkaConfig(logkafka_id, log_path, config, checkConfig)
      )
    ) {
      result: Future[CMCommandResult] =>
        result.map(cmr => toDisjunction(cmr.result))
    }
  }

  def deleteLogkafka(
                   clusterName: String,
                   logkafka_id: String,
                   log_path: String
                   ): Future[ApiError \/ ClusterContext] =
  {
    implicit val ec = apiExecutionContext
    withKafkaManagerActor(KMClusterCommandRequest(clusterName, CMDeleteLogkafka(logkafka_id, log_path))) {
      result: Future[CMCommandResult] =>
        result.map(cmr => toDisjunction(cmr.result))
    }
  }

  //--------------------Queries--------------------------
  def getClusterConfig(clusterName: String): Future[ApiError \/ ClusterConfig] = {
    tryWithKafkaManagerActor(KMGetClusterConfig(clusterName)) { result: KMClusterConfigResult =>
      result.result.get
    }
  }

  def getClusterContext(clusterName: String): Future[ApiError \/ ClusterContext] = {
    tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, CMGetClusterContext))(
      identity[ClusterContext]
    )
  }

  def getClusterList: Future[ApiError \/ KMClusterList] = {
    tryWithKafkaManagerActor(KMGetAllClusters)(identity[KMClusterList])
  }

  def getClusterView(clusterName: String): Future[ApiError \/ CMView] = {
    tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, CMGetView))(identity[CMView])
  }

  def getTopicList(clusterName: String): Future[ApiError \/ TopicList] = {
    tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, KSGetTopics))(identity[TopicList])
  }

  def getTopicListExtended(clusterName: String): Future[ApiError \/ TopicListExtended] = {
    val futureTopicIdentities = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, BVGetTopicIdentities))(
      identity[Map[String, TopicIdentity]])
    val futureTopicList = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, KSGetTopics))(identity[TopicList])
    val futureTopicToConsumerMap = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, BVGetTopicConsumerMap))(
      identity[Map[String, Iterable[(String, ConsumerType)]]])
    val futureTopicsReasgn = getTopicsUnderReassignment(clusterName)
    implicit val ec = apiExecutionContext
    for {
      errOrTi <- futureTopicIdentities
      errOrTl <- futureTopicList
      errOrTCm <- futureTopicToConsumerMap
      errOrRap <- futureTopicsReasgn
    } yield {
      for {
        ti <- errOrTi
        tl <- errOrTl
        tcm <- errOrTCm
        rap <- errOrRap
      } yield {
        TopicListExtended(tl.list.map(t => (t, ti.get(t))).sortBy(_._1), tcm, tl.deleteSet, rap, tl.clusterContext)
      }
    }
  }

  def getConsumerListExtended(clusterName: String): Future[ApiError \/ ConsumerListExtended] = {
    val futureConsumerIdentities = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, BVGetConsumerIdentities))(
      identity[Map[(String, ConsumerType), ConsumerIdentity]])
    val futureConsumerList = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, KSGetConsumers))(identity[ConsumerList])
    implicit val ec = apiExecutionContext
    for {
      errorOrCI <- futureConsumerIdentities
      errorOrCL <- futureConsumerList
    } yield {
      for {
        ci <- errorOrCI
        cl <- errorOrCL
      } yield {
        ConsumerListExtended(cl.list.map(c => ((c.name, c.consumerType), ci.get((c.name, c.consumerType)))), cl.clusterContext)
      }
    }
  }

  def getTopicsUnderReassignment(clusterName: String): Future[ApiError \/ IndexedSeq[String]] = {
    val futureReassignments = getReassignPartitions(clusterName)
    implicit val ec = apiExecutionContext
    futureReassignments.map {
      case -\/(e) => -\/(e)
      case \/-(rap) =>
        \/-(rap.map { asgn =>
          asgn.endTime.map(_ => IndexedSeq()).getOrElse{
            asgn.partitionsToBeReassigned.map { case (t, s) => t.topic}.toSet.toIndexedSeq
          }
        }.getOrElse{IndexedSeq()})
    }
  }

  def getBrokerList(clusterName: String): Future[ApiError \/ BrokerListExtended] = {
    val futureBrokerList= tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, KSGetBrokers))(identity[BrokerList])
    val futureBrokerMetrics = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, BVGetBrokerMetrics))(identity[Map[Int,BrokerMetrics]])
    implicit val ec = apiExecutionContext
    futureBrokerList.flatMap[ApiError \/ BrokerListExtended] { errOrBrokerList =>
      errOrBrokerList.fold ({
        err: ApiError => Future.successful(-\/(err))
      }, { bl =>
        for {
          errOrbm <- futureBrokerMetrics.recover[ApiError \/ Map[Int,BrokerMetrics]] { case t =>
            \/-(Map.empty)
          }
        } yield {
          val bm = errOrbm.toOption.getOrElse(Map.empty)
          \/-(
            BrokerListExtended(
              bl.list,
              bm,
              if(bm.isEmpty) None else Option(bm.values.foldLeft(BrokerMetrics.DEFAULT)((acc, m) => acc + m)),
              bl.clusterContext
            ))
        }
      })
    }
  }

  def getBrokersView(clusterName: String): Future[\/[ApiError, Map[Int, BVView]]] = {
    implicit val ec = apiExecutionContext

    tryWithKafkaManagerActor(
      KMClusterQueryRequest(
        clusterName,
        BVGetViews
      )
    )(identity[Map[Int, BVView]])
  }

  def getBrokerView(clusterName: String, brokerId: Int): Future[ApiError \/ BVView] = {
    val futureView = tryWithKafkaManagerActor(
      KMClusterQueryRequest(
        clusterName,
        BVGetView(brokerId)
      )
    )(identity[Option[BVView]])
    implicit val ec = apiExecutionContext
    futureView.flatMap[ApiError \/ BVView] { errOrView =>
      errOrView.fold(
      { err: ApiError =>
        Future.successful(-\/[ApiError](err))
      }, { viewOption: Option[BVView] =>
        viewOption.fold {
          Future.successful[ApiError \/ BVView](-\/(ApiError(s"Broker not found $brokerId for cluster $clusterName")))
        } { view =>
          Future.successful(\/-(view))
        }
      }
      )
    }
  }

  def getTopicIdentity(clusterName: String, topic: String): Future[ApiError \/ TopicIdentity] = {
    val futureCMTopicIdentity = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, CMGetTopicIdentity(topic)))(
      identity[Option[CMTopicIdentity]]
    )
    implicit val ec = apiExecutionContext
    futureCMTopicIdentity.map[ApiError \/ TopicIdentity] { errOrTI =>
      errOrTI.fold[ApiError \/ TopicIdentity](
      { err: ApiError =>
        -\/[ApiError](err)
      }, { tiOption: Option[CMTopicIdentity] =>
        tiOption.fold[ApiError \/ TopicIdentity] {
          -\/(ApiError(s"Topic not found $topic for cluster $clusterName"))
        } { cmTopicIdentity =>
          cmTopicIdentity.topicIdentity match {
            case scala.util.Failure(t) =>
              -\/[ApiError](t)
            case scala.util.Success(ti) =>
              \/-(ti)
          }
        }
      }
      )
    }
  }

  def getConsumersForTopic(clusterName: String, topic: String): Future[Option[Iterable[(String, ConsumerType)]]] = {
    val futureTopicConsumerMap = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, BVGetTopicConsumerMap))(
      identity[Map[String, Iterable[(String, ConsumerType)]]])
    implicit val ec = apiExecutionContext
    futureTopicConsumerMap.map[Option[Iterable[(String, ConsumerType)]]] { errOrTCM =>
      errOrTCM.fold[Option[Iterable[(String, ConsumerType)]]] (_ => None, _.get(topic))
    }
  }

  def getGeneratedAssignments(clusterName: String, topic: String): Future[ApiError \/ GeneratedPartitionAssignments] = {
    tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, CMGetGeneratedPartitionAssignments(topic)))(
      identity[GeneratedPartitionAssignments]
    )
  }

  private[this] def tryWithConsumerType[T](consumerType: String)(fn: ConsumerType => Future[ApiError \/ T])(implicit ec: ExecutionContext) : Future[ApiError \/ T] = {
    Future.successful[ApiError \/ ConsumerType] {
      ConsumerType.from(consumerType).fold[ApiError \/ ConsumerType](-\/(ApiError(s"Unknown consumer type : $consumerType"))){
        ct => \/-(ct)
      }
    }.flatMap { ctOrError =>
      ctOrError.fold(err => Future.successful[ApiError \/ T](-\/(err)), fn)
    }
  }

  def getConsumerIdentity(clusterName: String, consumer: String, consumerType: String): Future[ApiError \/ ConsumerIdentity] = {
    implicit val ec = apiExecutionContext
    val futureCMConsumerIdentity = tryWithConsumerType(consumerType) { ct =>
        tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, CMGetConsumerIdentity(consumer, ct)))(
          identity[CMConsumerIdentity]
        )
    }
    futureCMConsumerIdentity.map[ApiError \/ ConsumerIdentity] { errOrCI =>
      errOrCI.fold[ApiError \/ ConsumerIdentity](
      { err: ApiError =>
        -\/[ApiError](err)
      }, { ci: CMConsumerIdentity =>
          ci.consumerIdentity match {
            case scala.util.Failure(c) =>
              -\/[ApiError](c)
            case scala.util.Success(ci) =>
              \/-(ci)
          }
      })
    }
  }

  def getConsumedTopicState(clusterName: String, consumer: String, topic: String, consumerType: String): Future[ApiError \/ ConsumedTopicState] = {
    implicit val ec = apiExecutionContext
    val futureCMConsumedTopic = tryWithConsumerType(consumerType) { ct =>
      tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, CMGetConsumedTopicState(consumer, topic, ct)))(
        identity[CMConsumedTopic]
      )
    }
    futureCMConsumedTopic.map[ApiError \/ ConsumedTopicState] { errOrCT =>
      errOrCT.fold[ApiError \/ ConsumedTopicState](
      { err: ApiError =>
        -\/[ApiError](err)
      }, { cmConsumedTopic: CMConsumedTopic =>
        cmConsumedTopic.ctIdentity match {
          case scala.util.Failure(c) =>
            -\/[ApiError](c)
          case scala.util.Success(ci) =>
            \/-(ci)
        }
      }
      )
    }
  }

  def getTopicMetrics(clusterName: String, topic: String): Future[ApiError \/ Option[BrokerMetrics]] = {
    tryWithKafkaManagerActor(
      KMClusterQueryRequest(
        clusterName,
        BVGetTopicMetrics(topic)
      )
    ) { brokerMetrics: Option[BrokerMetrics] =>
      brokerMetrics
    }
  }

  def getPreferredLeaderElection(clusterName: String): Future[ApiError \/ Option[PreferredReplicaElection]] = {
    tryWithKafkaManagerActor(
      KMClusterQueryRequest(
        clusterName,
        KSGetPreferredLeaderElection
      )
    )(identity[Option[PreferredReplicaElection]])
  }

  def getReassignPartitions(clusterName: String): Future[ApiError \/ Option[ReassignPartitions]] = {
    tryWithKafkaManagerActor(
      KMClusterQueryRequest(
        clusterName,
        KSGetReassignPartition
      )
    )(identity[Option[ReassignPartitions]])
  }

  def topicListSortedByNumPartitions(tle: TopicListExtended): Seq[(String, Option[TopicIdentity])] = {
    def partition(tiOption: Option[TopicIdentity]): Int = {
      tiOption match {
        case Some(ti) => ti.partitions
        case None => 0
      }
    }
    val sortedByNumPartition = tle.list.sortWith{ (leftE, rightE) =>
      partition(leftE._2) > partition(rightE._2)
    }
    sortedByNumPartition
  }

  def getLogkafkaLogkafkaIdList(clusterName: String): Future[ApiError \/ LogkafkaLogkafkaIdList] = {
    tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, LKSGetLogkafkaLogkafkaIds))(identity[LogkafkaLogkafkaIdList])
  }

  def getLogkafkaListExtended(clusterName: String): Future[ApiError \/ LogkafkaListExtended] = {
    val futureLogkafkaIdentities = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, LKVGetLogkafkaIdentities))(identity[Map[String, LogkafkaIdentity]])
    val futureLogkafkaList = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, LKSGetLogkafkaLogkafkaIds))(identity[LogkafkaLogkafkaIdList])
    implicit val ec = apiExecutionContext
    for {
      errOrLi <- futureLogkafkaIdentities
      errOrLl <- futureLogkafkaList
    } yield {
      for {
        li <- errOrLi
        ll <- errOrLl
      } yield {
        LogkafkaListExtended(ll.list.map(l => (l, li.get(l))), ll.deleteSet)
      }
    }
  }

  def getLogkafkaIdentity(clusterName: String, logkafka_id: String): Future[ApiError \/ LogkafkaIdentity] = {
    val futureCMLogkafkaIdentity = tryWithKafkaManagerActor(KMClusterQueryRequest(clusterName, CMGetLogkafkaIdentity(logkafka_id)))(
      identity[Option[CMLogkafkaIdentity]]
    )
    implicit val ec = apiExecutionContext
    futureCMLogkafkaIdentity.map[ApiError \/ LogkafkaIdentity] { errOrLI =>
      errOrLI.fold[ApiError \/ LogkafkaIdentity](
      { err: ApiError =>
        -\/[ApiError](err)
      }, { liOption: Option[CMLogkafkaIdentity] =>
        liOption.fold[ApiError \/ LogkafkaIdentity] {
          -\/(ApiError(s"Logkafka not found $logkafka_id for cluster $clusterName"))
        } { cmLogkafkaIdentity =>
          cmLogkafkaIdentity.logkafkaIdentity match {
            case scala.util.Failure(l) =>
              -\/[ApiError](l)
            case scala.util.Success(li) =>
              \/-(li)
          }
        }
      }
      )
    }
  }

  def initialiseSchedulePreferredLeaderElection(): Unit = {
    implicit val ec = apiExecutionContext
    implicit val formats = org.json4s.DefaultFormats

    var temp: Map[String, Int] = Map.empty
    val x = system.actorSelection(kafkaManagerActor).ask(KSGetScheduleLeaderElection)
    x.foreach { schedule =>
      temp = parse(schedule.toString).extract[Map[String, Int]]
      for ((cluster, timeInterval) <- temp) {
        schedulePreferredLeaderElection(cluster, Set(), timeInterval)
      }
    }
  }

  /* Contains a key for each cluster (by its name) which has preferred leader election scheduled
  * Value of each key is a 2-tuple where
  * * first element is the scheduler's cancellable object - required for cancelling the schedule
  * * second element is the time interval for scheduling (in minutes) - required for storing in ZK
  */
  var pleCancellable : Map[String, (Option[Cancellable], Int)] = Map.empty
  initialiseSchedulePreferredLeaderElection()

}

redpajama_set_name: RedPajamaGithub


text:
Q: java.util.logging: Remove date time line I currently have this code in my main().
Logger logger = Logger.getLogger("My Log");
FileHandler fh;
logger.addHandler(fh);
SimpleFormatter formatter = new SimpleFormatter();
fh.setFormatter(formatter);
logger.info("Test");

This outputs 2 lines. One with datetime, and the other is test. I read many other questions on this site, but i do not get how to implement it to my current code.
UPDATE: I have created a new file called logging.properties and updated Logger logger = Logger.getLogger("logging.properties");
Inside the file i inputted these values
handlers= java.util.logging.ConsoleHandler
.level= INFO
java.util.logging.ConsoleHandler.level = INFO
java.util.logging.ConsoleHandler.formatter = java.util.logging.SimpleFormatter
javax.jms.connection.level = INFO
java.util.logging.SimpleFormatter.format=[%1$tF %1$tr] %3$s %4$s:  %5$s %n

But when i run i still see the date line. What am i missing?

A: You need to add a file called logging.properties to your Java project. In there, you can define the format to be used:
java.util.logging.SimpleFormatter.format=[%1$tF %1$tr] %3$s %4$s:  %5$s %n

The documentation here explains what the %... specifiers mean: http://docs.oracle.com/javase/7/docs/api/java/util/Formatter.html#syntax

A: If you want to do it programatically, you could do 2 things:

*

*Implement a custom formatter and set to handler: fh.setFormatter(new MyCustomFormatter()); See http://examples.javacodegeeks.com/core-java/util/logging/create-custom-formatter-for-logger-handler/

*If you want only 1 line, the workaround is explained here:
java.util.logging: how to suppress date line

redpajama_set_name: RedPajamaStackExchange


The gas and electric industries are unique, complex, and increasingly interrelated. This one-day seminar presents a high-level overview of the business of natural gas and electricity and covers topics including customers, physical systems, operations, regulation, and the marketplace. For those new to the energy industry as well as veterans who aren't quite sure how all the pieces fit together, Gas and Electric Industry Basics – Condensed version will help participants understand the basic principles that drive the gas and electric businesses.


text:
\section{Introduction}

Hard exclusive electroproduction of mesons from nucleons has become
a field of growing interest.
The  recent progress in the QCD analysis of such processes
is based on a factorization theorem \cite{CFS97}
(see also \cite{Rad97,Hood98,MPW98a,Vanderh98a})
valid in the case of longitudinally polarized photons:
at large photon virtualities, $Q^2 \gg \Lambda_{\rm{QCD}}^2$,
the underlying photon-parton sub-processes are dominated by
short distances and,
hence,
 can be calculated perturbatively.
On the other hand, all information on the long distance dynamics of
quarks and gluons can be collected in the distribution
amplitude of the produced meson  and in generalized
(skewed \cite{Rad97,Ji98} or double \cite{Rad97})
parton distribution functions of the nucleon.
As a consequence, experimental investigations
of meson production
processes allow to probe details of the quark-gluon dynamics in
nucleons beyond our current knowledge obtained
mainly from inclusive high-energy processes.
Ongoing and planned measurements at DESY
(HERA \cite{HERA}, HERMES \cite{HERMES}),
CERN (COMPASS \cite{COMPASS}), and Jefferson Lab \cite{TJNAF}
are therefore of great interest.

In
our
previous work
\cite{MPW98a,MPW98b}  we have derived
the leading order QCD production amplitudes for neutral
pseudoscalar mesons and vector mesons.
The corresponding cross sections have been discussed within
a specific model
\cite{Rad98a}
 for

double
parton distributions.
In this paper  we consider hard electroproduction of charged pions.
Interest in this process arises in particular from the fact that
it receives significant contributions from two
production mechanisms:
either (i) the  photon interacts with a quark  from the nucleon
which, after hard gluon exchange, combines with a second quark
of the target to the final pion, or (ii) the pion is produced
from the meson cloud of the nucleon. The latter mechanism is often referred
to as the pion pole contribution.
In meson-cloud models of the nucleon  the second mechanism
directly involves the electromagnetic form factor $F_{\pi}(Q^2)$ of
the pion (see e.g. \cite{Speth} and references therein).
Pion electroproduction has therefore often
been considered as a generic process to determine
$F_{\pi}(Q^2)$ (see e.g. \cite{bebek}).
However, as discussed in ref.\cite{Carlson90},
there may be significant uncertainties
due to  type (i) contributions.
The latter have been modeled in ref.\cite{Carlson90}
in terms of  hard gluon exchange diagrams with nonperturbative
factors estimated
by an overlap of light-cone wave functions
with Chernyak-Zhitnitsky type distribution amplitudes \cite{CZ}.

In this work   we address
questions similar to those raised
in \cite{Carlson90}.
The distinctive feature of our approach is a systematic use
of perturbative QCD (PQCD)
factorization  \cite{CFS97,Rad97}.
Within a consistent PQCD analysis
all necessary information about the nucleon
structure  is contained in   quark
double
distributions.
The latter
are constrained through the behavior of ordinary quark
distribution functions, as well as through sum rules
and symmetry properties (for a discussion see
\cite{MPW98a,Rad98c}).
In particular we
show that actually both production mechanisms, (i) and (ii), can
be formulated in terms of
quark double distribution functions
of the nucleon, i.e.  the existence
of the pion pole contribution fits into
the
PQCD factorization framework.

After modeling the involved double distribution functions   we
calculate the differential $\pi^+$ production cross section
in the region of small momentum transfers, $-t < 0.5$ GeV$^2$.
We find that both mechanisms,
(i) and (ii), contribute significantly.
While one pion exchange dominates at intermediate
values of Bjorken-$x$, perturbative quark exchange
is relevant mainly at small $x_{\rm Bj}$.

The paper is organized as follows: in Sec.2 we present the
amplitudes for charged pion electroproduction. Pseudovector and pseudoscalar
parts of the relevant double distribution functions
are associated in Sec.3 and 4 with the
perturbative  and pion pole contribution, respectively.
Results are presented in Sec.5. Finally we
summarize.

\section{Amplitudes}

The calculation of meson electroproduction amplitudes
is based on   PQCD factorization  \cite{CFS97}.
It applies to incident longitudinally polarized photons
with large space-like momenta, $Q^2 = -q^2 \gg \Lambda_{\rm{QCD}}^2$,
and moderate momentum transfer  to the nucleon target,
$|t|\,\lsim  \Lambda_{\rm{QCD}}^2$.
Using the techniques  outlined in refs.\cite{CFS97,Rad97,MPW98a,MPW98b}
gives the following
result for the $\pi^+$ virtual photoproduction amplitude
in leading twist, leading order $\alpha_s$ accuracy:
\begin{eqnarray}
\hspace*{-1cm}
{\cal A}_{\pi^{+}}
&=& i \frac{g^2 C_F}{4 N_c} \frac{f_{\pi}}{Q}
           \frac{\bar{N}(P',S') \gamma_5 \hat{n} N(P,S)}{{\bar P} \cdot n}
           \int_0^1 d\tau \frac{\Phi_{\pi}(\tau)}{\tau \bar{\tau}}
          \int d\,[x,y] \,
\nonumber \\
&&
\hspace*{-1.5cm}
\times  \left[ (e_d\, \Delta F^{du} - e_u\,\Delta {\bar F}^{du})
   \frac{{\bar \omega}}{x + 2 y + x \bar{\omega} - i \epsilon} -
    (e_u \, \Delta F^{du} - e_d\, \Delta {\bar F}^{du})
    \frac{{\bar \omega}}{x + 2 y - x \bar{\omega} - i \epsilon} \right]
    \nonumber \\
&&\hspace*{-1.5cm}
-i \frac{g^2 C_F}{2 N_c} \frac{f_{\pi}}{Q}
           \frac{\bar{N}(P',S') \gamma_5  N(P,S)}{2 M}
           \int_0^1 d\tau \frac{\Phi_{\pi}(\tau)}{\tau \bar{\tau}}
          \int d\, [x,y]
\nonumber \\
&&\hspace*{-1.5cm}
\times  \left[ (e_d\, \Delta K^{du} - e_u\,\Delta {\bar K}^{du})
   \frac{{1}}{x + 2 y + x \bar{\omega} - i \epsilon} -
    (e_u \, \Delta K^{du} - e_d\, \Delta {\bar K}^{du})
    \frac{{1}}{x + 2 y - x \bar{\omega} - i \epsilon} \right].
\nonumber \\
\label{eq:vecm_quark_diag}
\end{eqnarray}
Here the notation $\int d\, [x,y] \dots = \int_0^1 dx \int_0^{\bar x}
dy$ \dots,  with $\bar x = 1-x$ is used.
$N(P,S)$ and $\bar N(P',S')$  are the  Dirac spinors
of the initial and scattered nucleon, respectively,
with the corresponding four-momenta  $P$, $P'$
and spins $S$, $S'$. $M$ stands for the nucleon mass.
The average nucleon momentum is denoted by $\bar P = (P + P')/2$,
and the momentum transfer is  $r = P-P'$
with $t = r^2$.
The produced  meson carries the four-momentum
$q'$ and $\bar q = (q + q')/2$.
Furthermore, we have introduced the variable
$\bar \omega = 2 \bar q \cdot \bar P/ (- \bar q^2)$.
Finally, $n$ is  a light-like vector with
$n\cdot a = a^+ = a^0+ a^3$ \
for any vector $a$, and $\hat n = \gamma_{\mu} n^{\mu}$.

The long distance dynamics of the
pion in the final state
is contained in the decay
constant $f_{\pi} = 133$ MeV and the distribution amplitude
$\Phi_{\pi}(\tau)$ \cite{ER78,BL79}:
\begin{equation} \label{eq:meson_distr_VL}
\langle \pi^+ (q') |\, \bar{\psi}_u(x) \gamma_5\hat n \psi_d(y)
\, | 0 \rangle
  =  - i f_{\pi} \,(q'\cdot n) \,\int_0^1  \,
    \Phi_{\pi}(\tau) \,
    e^{i q' \cdot \, (\tau x + \bar{\tau} y)} \, d\tau \, ,
\end{equation}
where quark fields $\psi$ with proper flavor quantum numbers
enter.

The nucleon part of the amplitude (\ref{eq:vecm_quark_diag}) is determined
by double distribution functions
$\Delta F^{du}$ and $\Delta K^{du}$
which are nondiagonal in flavor.
They are defined as matrix elements of a nonlocal quark
operator sandwiched between proton and neutron states:
\begin{eqnarray} \label{eq:dd_quark_ud}
&&\hspace{-1.5cm}
\left \langle n(P',S')\right|
\bar\psi_d(0) \gamma_5\hat n \left[0,z\right] \psi_u(z)
\left|p(P,S)\right \rangle_{z^2 = 0}
=
\nonumber \\
&&\bar N(P',S')\,\gamma_5 \hat n \,N(P,S)
\int d\, [x,y]
\left(
e^{-i x (P\cdot z) - i y(r\cdot z)} \Delta F^{du}
+
e^{i x (P\cdot z) - i \bar y(r\cdot z)} \Delta \bar F^{du}
\right)
\nonumber \\
&-&
\bar N(P',S')\,\gamma_5 \,N(P,S) \,\frac{r \cdot n}{2 M}
\int d\, [x,y]
\left(
e^{-i x (P\cdot z) - i y(r\cdot z)} \Delta K^{du}
+
e^{i x (P\cdot z) - i \bar y(r\cdot z)} \Delta \bar K^{du}
\right).
\nonumber \\
\end{eqnarray}
Here, as in (\ref{eq:vecm_quark_diag}),  the dependence of
the double distribution functions on $x, y$ and $t$ has been
suppressed.
The up- and down-quark fields in (\ref{eq:dd_quark_ud})
are separated by a light-like distance $z\sim n$.
Gauge invariance is guaranteed by the
path-ordered exponential
$$[0,z] = \
{\cal P} \exp [ -i g z_{\mu}
\-\int_0^1  A^{\mu}(z \lambda) \, d \lambda]$$
which reduces to
$1$
in  axial gauge $n\cdot A=0$
($g$ stands for the strong coupling constant and
$A^{\mu}$ denotes the gluon field).

\section{Pseudovector contribution}

The production amplitude ${\cal A}_{\pi^+}$ contains
a pseudovector and pseudoscalar contribution
proportional to  $\bar N(P',S')\,\gamma_5 \hat n \,N(P,S)$
and $\bar N(P',S')\,\gamma_5 \,N(P,S)$, respectively.
We rewrite the pseudovector part according to isospin
relations which connect  flavor non-diagonal double distribution
functions $\Delta F^{du}$ with flavor diagonal ones.
In particular we use \cite{MPW98b}:
\begin{equation}
\left \langle n \right| {\hat O^{du}}(z) \left|p\right \rangle =
\left \langle p\right| {\hat O^{uu}}(z) \left|p\right \rangle -
\left \langle p\right| {\hat O^{dd}}(z) \left|p\right \rangle,
\end{equation}
with
\begin{equation}
\left.{\hat O}^{q\,q^\prime}(z) =
\bar\psi_q(0) \gamma_5\hat n \left[0,z\right]
\psi_{q^\prime}(z)\right|_{z^2 = 0}.
\end{equation}

In terms of flavor diagonal polarized double distribution
functions the pseudovector part of the amplitude
(\ref{eq:vecm_quark_diag}) reads:
\begin{eqnarray}
{\cal A}_{\pi^{+}}
&=& - i \frac{g^2 C_F}{8 N_c} \frac{f_{\pi}}{Q}
           \frac{\bar{N}(P',S') \gamma_5 \hat{n} N(P,S)}{{\bar P} \cdot n}
           \int_0^1 d\tau \frac{\Phi_{\pi}(\tau)}{\tau \bar{\tau}}
           \int d\, [x,y] \Bigg{\{}
\nonumber \\
&&
(e_u\! - \!e_d)
   \left[(\Delta F^u \!+\! \Delta {\bar F}^u) -  (
          \Delta F^d\! +\! \Delta {\bar F}^d)\right]
 \left(  \frac{{\bar \omega}}
{x\! + \!2 y\! +\! x \bar{\omega}\! - \!i \epsilon}\! +\!
\frac{{\bar \omega}}
{x \!+\! 2 y\! - \!x \bar{\omega}\! - \!i \epsilon} \right)
\nonumber \\
&-&
(e_u \!+ \!e_d)
   \left[(\Delta F^u \!- \!\Delta {\bar F}^u) -
         (\Delta F^d \!-\! \Delta {\bar F}^d)\right]
 \left(\frac{{\bar \omega}}
{x\! + \!2 y\! +\! x \bar{\omega} \!- \!i \epsilon} -
         \frac{{\bar \omega}}
{x \!+\! 2 y\! - \!x \bar{\omega} \!- \!i \epsilon} \right)
\Bigg{\}}.
\nonumber \\
\label{eq:piplus_quark_diag}
\end{eqnarray}
Here $e_u$ and $e_d$ denote the electromagnetic charges of
the up- and down-quarks.
Note that ${\cal A}_{\pi^{+}}$
has a structure similar to the $\rho^+$ production
amplitude derived in \cite{MPW98b}:
up to prefactors the  $\rho^+$ amplitude  can be  obtained
from (\ref{eq:piplus_quark_diag}) by replacing polarized quark
double distribution functions  by unpolarized ones,
with proper account of
their opposite charge conjugation properties.

In order to obtain a numerical estimate  for the
pseudovector contribution (\ref{eq:piplus_quark_diag})
to $\pi^+$ production models for the involved
double distributions $\Delta F$ have to be constructed.
We are guided here by the appropriate forward limit of
double distributions, their symmetry properties,
and sum rules which relate them to nucleon form factors.
Following refs. \cite{Rad98b} and \cite{Rad98c} we use:
\begin{eqnarray} \label{eq:f_models}
A: \quad  \Delta F(x,y;t) & = & h(x,y) \, \Delta q(x) \, f(t),
\nonumber \\
\nonumber \\
B: \quad \Delta F(x,y;t) & = &
h(x,y) \, \Delta q(x) \,
\exp\left[\frac{t}{\Lambda^2} \, \frac{y(1-x-y)}{x(1-x)}\right].
\end{eqnarray}
Here $h(x,y) = 6 \, y \,(1-x-y)/(1-x)^3$,
and $\Delta q(x)$ denotes  the corresponding ordinary polarized
quark distribution.
  In numerical
  calculations presented in this paper we have used Gehrman-Stirling LO set-A
  parametrizations of polarized quark distributions \cite{Gehrmann96}.
Furthermore,
we have used the one loop expression for the running coupling constant
with $\Lambda_{\rm QCD} = 200$ MeV.
Note  that
in addition to
the above mentioned constraints,
the models (\ref{eq:f_models})
are consistent with  asymptotic solutions of evolution equations
for double distributions
and satisfy the symmetries of the latter \cite{MPW98a}.
The $t$-dependence
of double distributions in  model $A$ is governed by the form factor
$f(t)$ which, like in \cite{MPW98a}, is taken to be equal to
the nucleon pseudovector form factor \cite{Weise88}:
\begin{equation}\label{eq:ffac_model_1}
f(t) = \frac{1}{(1-t/\Lambda^2)^2}.
\end{equation}
The exponential dependence of $\Delta F$ on
the momentum transfer $t$ in model $B$ can be
obtained from a simple field theoretical
investigation of double distribution
functions outlined in \cite{Rad98c,Rad98b}.
For both models the scale $\Lambda$ has been  fixed at $1$ GeV.
With this choice   model $B$
reproduces the nucleon axial form factor up to momentum transfers
$t \approx - 0.5$ GeV$^2$.
Another important feature of model $B$ is, that the small-$x$
behavior of $\Delta F$ becomes less and less singular as $|t|$ increases.
Such a behavior is expected since, like in the case of nucleon
form factors, large momentum transfers filter out the minimal
contribution to the Fock space wave function of the target.

\section{Pseudoscalar contribution}

As explained in ref.\cite{Rad98b}  the double
distributions $\Delta F(x,y;t)$ can be related in the forward
limit, $t = 0$, to ordinary polarized quark densities $\Delta q(x)$.
Therefore, it  is possible to construct plausible models for
$\Delta F$,
although little is known about their shape from first principles.
On the other hand, for the double distributions $\Delta K$
no experimental information on  corresponding forward densities is available
which makes modeling in this case more difficult.
Nevertheless, some intuition about their magnitude has been obtained from
model calculations \cite{Models}.
Still,
contributions of the so-called $K$-terms
to hard exclusive meson production have  been neglected
so far.
In the present case of $\pi^+$
production the situation is  different. As
we will show,  the pseudoscalar piece of the amplitude ${\cal A}_{\pi^+}$
can be associated with the virtual photoproduction of pions from
the nucleon  pion cloud.
Due to the small pion mass  this mechanism is expected to be
important, especially at small  momentum transfers $t$.

We are interested in evaluating the contribution to the
nucleon matrix element
(\ref{eq:dd_quark_ud}) arising from a  situation when a quark-antiquark
pair with  low invariant mass propagates  in the
$t$-channel, such that it becomes close to a  $\pi^+$ bound state.
To construct the corresponding double distribution function we
introduce an effective Lagrangian describing a pseudovector pion-nucleon
interaction \cite{Weise88}:
\begin{equation}
{\cal L}_{\pi NN} =
\frac{g_{\pi NN}}{2 M}
\left( \phi(x) \gamma_{\mu} \gamma_5 \vec\tau \phi(x)\right)
\cdot \left(\partial^{\mu} \vec \varphi(x)\right).
\end{equation}
The nucleon and pion fields are denoted by
$\phi$ and $\vec \varphi$, respectively,
$\vec \tau$ are  the common isospin matrices,
and $g_{\pi N N}$ is the pion-nucleon coupling constant.
Evaluating the matrix element (\ref{eq:dd_quark_ud}) to
first order in $g_{\pi NN}$ leads to:
\begin{eqnarray}
&&
\hspace*{-1cm}
\left \langle n(P',S')\right|
\bar\psi_d(0) \gamma_5\hat n \left[0,z \right] \psi_u(z)
\left|p(P,S)\right \rangle_{z^2 = 0}
= \sqrt{2} g_{\pi NN} \,
\bar N(P',S')\,\gamma_5 \,N(P,S)
\nonumber \\
&&
\times \int d^4 x  \, e^{- i r \cdot x}
\left \langle 0 \right|
T[
\bar\psi_d(0) \gamma_5\hat n \left[0,z \right] \psi_u(z)
\, \phi^+(x) ]
\left| 0  \right \rangle_{z^2 = 0} \, ,
\label{eq:repl1}
\end{eqnarray}
where $\varphi^+ = \frac{1}{\sqrt{2}}(\varphi_1+i\, \varphi_2)$ is the pion
field associated with the  $\pi^+$ meson. The leading-twist contribution
to the matrix element (\ref{eq:repl1}) arises from the
region of phase space where the quark fields are localized
near the light-cone, $z^+ \sim 0$.
Inserting a full set of intermediate pion states between the pion
field and the non-local quark operator gives:
\begin{eqnarray} \label{eq:ps_contr}
&&
\hspace*{-1cm}
\left \langle n(P',S')\right|
\bar\psi_d(0) \gamma_5\hat n \left[0,z\right] \psi_u(z)
\left|p(P,S)\right \rangle_{z^2 = 0}
\nonumber \\
&&
=
N(P',S') \gamma_5 N(P,S)\,
\frac{- i \sqrt{2} g_{\pi NN}}{m_\pi^2 - t}\,
\langle 0 |
\bar\psi_d(0) \gamma_5\hat n \left[0,z\right] \psi_u(z)
|\pi^+(r)\rangle_{z^2 = 0}.
\end{eqnarray}
Using the pion distribution amplitude (\ref{eq:meson_distr_VL}) and
comparing with the definition of the
pseudoscalar double distribution functions in eq.(\ref{eq:dd_quark_ud})
yields:
\begin{eqnarray}\label{eq:mel_ps_final}
\left(\Delta K^{du}  + \Delta \bar K^{du}\right)(x,y,t\approx 0)
&=&
-\frac{2 \sqrt{2} f_{\pi} M g_{\pi NN}}{m_\pi^2 - t}
\,\delta (x) \,\Phi_{\pi}(y).
\end{eqnarray}
As this contribution can be expressed entirely in terms of $\Delta K$
there is no danger of double counting if we  add this term to the
contribution arising from a model for $\Delta F$, as discussed in the
previous section.

One can express the $t$-channel one pion exchange
contribution also in the form:
\begin{equation}\label{eq:dir_pion_explicit}
{\cal A}_{\pi^{+}} =
-\sqrt{2} g_{\pi NN}
\frac{N(P',S') \gamma_5 N(P,S)}{m_\pi^2 - t}\,
\,\varepsilon_L \cdot (q^\prime + r) \,F_{\pi^+}(Q^2),
\end{equation}
where
\begin{equation}\label{eq:e_L_def}
\varepsilon_L = \frac{i}{Q}\, (q^\prime + r)
\end{equation}
stands for  the polarization vector of the longitudinally polarized photon,
and
\begin{equation}\label{eq:F_pi_Q2}
F_{\pi^+}(Q^2) = (e_u-e_d)\, \frac{g^2 C_F}{2 N_c} \frac{f_\pi^2}{Q^2}
\left(\int_0^1 du\, \frac{\Phi(u)}{u}\right)^2
\end{equation}
is the leading QCD contribution
to the $\pi^+$ electromagnetic form factor at
large $Q^2$ \cite{ER78,BL79}.

To twist-2 accuracy, i.e. neglecting terms of  order $t/Q^2$
and $m_\pi^2/Q^2$, the amplitude
(\ref{eq:dir_pion_explicit}) is explicitly gauge invariant.
This is guaranteed by the factor $\varepsilon_L \cdot (q^\prime + r)$
which arises from  the hard photon-quark interaction.
In the one pion exchange approximation  the
difference $\Delta K^{du} - \Delta \bar K^{du}$ decouples from the considered
production process.
Finally, note that in the case of $\pi^0$ production
a $t$-channel exchange contribution similar to
(\ref{eq:mel_ps_final}) appears in the production
amplitude with a zero coefficient from the difference
of two equal quark charges.

To this end some comments regarding  one pion $t$-channel exchange,
as used above, are in order.
In general one expects that at large center-of-mass energies
any $t$-channel exchange of an hadronic state should be
replaced by the corresponding Regge trajectory.
Therefore, at small values of Bjorken-$x$ additional contributions
to the pseudoscalar part of the production amplitude
${\cal A}_{\pi^+}$ could occur.
Once the pion pole contribution is replaced by  reggeized pion (and rho)
exchange \cite{Guidal97} the relation to the pion
electromagnetic form factor seems to be lost
(see, however, ref.\cite{Vanderh98b} for an attempt to interpret pion
electroproduction data in the framework of a  Regge exchange  model).
We restrict ourselves in this work to the
simplest model for the pseudoscalar part of the production
amplitude in the hope that it contains the dominant
contribution at small momentum transfers, i.e. close to the
pion pole.

\section{Results}

In Figs.1 and 2 we present results for the
cross section  of  exclusive $\pi^+$ production
in the scattering of longitudinally polarized
photons from nucleons  at $Q^2 = 10$ GeV$^2$.
Up to logarithmic corrections the  cross
section is proportional to $1/Q^6$ as one can infer from
eq.(\ref{eq:vecm_quark_diag}).

In Fig.1 we show the differential production cross section
for the minimal kinematically
allowed value of $t = t_{\rm min} = -x_{\rm Bj}^2 M^2/(1-x_{\rm Bj})$.
We find that the pseudovector contribution (\ref{eq:piplus_quark_diag})
dominates at small values of the Bjorken variable $x_{\rm Bj}$.
Its $x_{\rm Bj}$-dependence reflects the behavior of
the relevant
forward parton distributions (\ref{eq:f_models}).
At $x\sim 0.3$ the cross section is governed by the pseudoscalar
pion cloud contribution
(\ref{eq:ps_contr}).

While models $A$ and $B$ give qualitatively similar cross sections
at small momentum transfers,  they start to differ for increasing  $|t|$.
This is illustrated in Fig.2
where we present results for the differential production
cross section taken at $t = - 0.4$ GeV$^2$.
The maximum value of Bjorken-$x$ is determined by the condition
$|t| \geq |t_{\rm min}|$.
For model $A$ the pseudovector contribution decreases slowly
with  decreasing $x_{\rm Bj}$ and dominates the cross
section at $x_{\rm Bj} \lsim  0.1$.
At $x_{\rm Bj} \sim x_{\rm max}$ the total
pion production cross section is about half as large as
obtained from the pion cloud contribution alone due to
strong interference effects.
On the other hand, in model $B$ the $x_{Bj}$-dependence of
the pseudovector contribution is similar to the one
from the pion cloud.
Also, contrary to the previous case, at
$x_{\rm Bj} \sim x_{\rm max}$ the production cross
section is about  20\% larger than
obtained from the pion cloud alone.

Finally we compare  in Fig.3  to the leading twist
production cross section for $\pi^0$ mesons
which has been investigated  in \cite{MPW98a}.
At $t=t_{\rm min}$ the pseudovector contribution to $\pi^+$
production is about a factor $5$--$10$ larger as compared to
the $\pi^0$ case.
At $x_{\rm Bj}\sim 0.3$ large contributions from the pion cloud,
which are absent in $\pi^0$ production, add further
to the cross section ratio.

It should be emphasized that the dominance of
the hard gluon exchange processes considered in this
paper can only be ensured for sufficiently large values of $Q^2$.
Just like in the case of the pion form factor,
competing contributions  are provided by
soft mechanisms  which, in this case, correspond
to the nonperturbative overlap of three hadronic wave functions.
The interplay of soft and hard interaction mechanisms
has been studied  recently  for the pion form factor in a
QCD sum rule related model \cite{SJR98}.
Guided by the results of this investigation we expect
the  dominance of
hard gluon exchange for values of
$Q^2$ above 10 GeV$^2$.

\section{Summary}

We have discussed the hard exclusive electroproduction of $\pi^+$ mesons
from nucleons.
For longitudinally polarized photons one can
express the corresponding photoproduction amplitude
to leading twist, leading order $\alpha_s$ accuracy
in terms of quark double distribution functions of the nucleon.
We find that the amplitude receives contributions
from pseudovector and pseudoscalar pieces.
The pseudovector contribution is determined by quark
double distributions with well known
properties in the forward limit, as given by
ordinary polarized quark distribution functions.
However, for the pseudoscalar part such information
is not available.
In this work it has been modeled in terms of an interaction of
the virtual photon with the nucleon pion cloud.
As a consequence, the related double distribution is determined
by the pion distribution amplitude.
The corresponding contribution to the pion production
amplitude turns out to be proportional to the pion form factor.

We have found that both,  pseudovector and pseudoscalar
terms
should be included if one considers the whole $0 <x_{\rm Bj} <1$ range.
In particular,
we have observed
that the pseudovector part dominates at small
$x_{\rm Bj} < 0.1$.
On the other hand, the pion cloud contribution controls the
$\pi^+$ production
at small $t \sim t_{\rm min}$ and  $0.1 < x_{\rm Bj} < 0.4$.
At larger values of $|t| \sim 0.4$ GeV$^2$ the relative weight
of pseudoscalar and pseudovector contributions
has been found to be sensitive to details of the
relevant
double
distribution functions.
Further investigations of double distribution functions,
as well as experimental data on exclusive meson production
are certainly needed.

\bigskip

\noindent
{\bf Acknowledgments:}

\noindent
G.P. would like to thank the SLAC theory group for their hospitality.
Furthermore, we would like to acknowledge discussions with
V. Burkert, S.J. Brodsky, L. Frankfurt, M.V. Polyakov and P. Stoler.
The  work of A.R.  was supported by the
US Department of Energy under contract
DE-AC05-84ER40150.
\\
\noindent
After this paper has been completed, we learned about an
independent calculation of $\pi^+$ electroproduction
which uses skewed parton distributions computed in a
chiral quark soliton model
\cite{Bochum1}.
The corresponding results turn out to be quite similar to ours.
We thank the authors of \cite{Bochum1}
for discussions and an exchange of results.

redpajama_set_name: RedPajamaArXiv


A Possible Solar Storm in 2012?
Not long ago, a report quoting NASA scientists was issued by the National Academy of Science, the highest scientific advisory body to the United States Congress. The report said something that used to be unthinkable: our own sun could have a temper tantrum that would unleash a tsunami of radiation from outer space, potentially wiping out our space satellites, wiping out power lines, and leaving entire continents without power.
This means that your refrigerator and freezer wouldn't have any electricity, leaving food to rot and creating food shortages around the world. Communication systems and power systems worldwide could be wiped out for months at a time. Something this large would cause not just a localized but rather a continent-wide power outage.
An event of this magnitude may seem hard to grasp, but just remember that every eleven years the sun flips with respect to its magnetic field (North becomes South and South becomes North) and releases a burst of radiation directed toward outer space, including Earth, and that this burst could endanger our communication systems. We forget that the space age is very young; we've only had a few of these sunspot cycles to worry about, and for the most part we have dodged the bullet.
Most of the radiation that has been emitted over previous cycles has missed the earth (thank goodness), but in 1859 there was a solar storm which paralyzed our telegraph wires and was the largest geomagnetic storm in recorded history. Back then we only had telegraph, and our telegraph wires were quite sturdy. From the damage done we can calculate how massive the solar storm was, and it was a whopper! In fact, if that solar storm of 1859 were to hit us today it would cause (according to the report) trillions of dollars in property damage and hundreds of billions of dollars in damage for each of the countries that were affected. Food riots would break out, governments could collapse...and there would be no cavalry coming to the rescue because the cavalry would also lose its power and ability to communicate. Transformers could get wiped out, meaning that whole continents would not have electrical power. We'd essentially be back to the way we were in 1859. Gone too would be the satellites controlling most of our luxuries today, including communication, television, radio, and the Internet where you are reading this very blog.
This is the nightmare scenario. So when might this happen? You guessed it: the year 2012, the next peaking of our sun's solar cycle. The report, however, is careful to say that this may in fact not happen. It's inevitable, however, that at some time or another it will happen again. Think about Katrina: prior to 2005, a lot of people knew that Katrina-type hurricanes could wipe out the levees in New Orleans. It was a well-known fact, but nobody did anything because everyone said "Let the next Administration—let the next sucker deal with it." Well, unfortunately the next sucker was us, and we had to foot the multi-billion dollar bill of Katrina. Now scientists are saying that a possible Katrina from outer space—potentially many, many times more disastrous—could hit the earth, and it's inevitable that it will happen.
So why not reinforce our satellites—and our power systems—today?
Much as with Katrina, it pays to do certain things now because we do face the inevitable. For example, even though satellites currently in outer space cannot be reinforced, the next generation of satellites should be reinforced to withstand the full brunt of the solar winds and solar flares. Also, on Earth, power stations and especially transformers should be insulated and reinforced so as to be able to withstand radiation coming from outer space. This is not rocket science; we know how to do it and the technology is already there—it's just a question of political will. Some people say "Nah...I mean, it might happen in a few decades, but not on my watch." Well, it will happen, and when it does...watch out!
For now, let's hope we dodge another bullet the next time the sun decides to throw a temper tantrum.


text:
/**
@module ember
@submodule ember-template-compiler
*/

import isEnabled from 'ember-metal/features';
import { assign } from 'ember-metal/merge';
import defaultPlugins from 'ember-template-compiler/plugins';

/**
  @private
  @property compileOptions
*/
export default function(_options) {
  var disableComponentGeneration = true;
  if (isEnabled('ember-htmlbars-component-generation')) {
    disableComponentGeneration = false;
  }

  let options;
  // When calling `Ember.Handlebars.compile()` a second argument of `true`
  // had a special meaning (long since lost), this just gaurds against
  // `options` being true, and causing an error during compilation.
  if (_options === true) {
    options = {};
  } else {
    options = assign({}, _options);
  }

  options.disableComponentGeneration = disableComponentGeneration;

  let plugins = {
    ast: defaultPlugins.ast.slice()
  };

  if (options.plugins && options.plugins.ast) {
    plugins.ast = plugins.ast.concat(options.plugins.ast);
  }
  options.plugins = plugins;

  options.buildMeta = function buildMeta(program) {
    return {
      topLevel: detectTopLevel(program),
      revision: 'Ember@VERSION_STRING_PLACEHOLDER',
      loc: program.loc,
      moduleName: options.moduleName
    };
  };

  return options;
}

function detectTopLevel(program) {
  let { loc, body } = program;
  if (!loc || loc.start.line !== 1 || loc.start.column !== 0) { return null; }

  let lastComponentNode;
  let lastIndex;
  let nodeCount = 0;

  for (let i = 0, l = body.length; i < l; i++) {
    let curr = body[i];

    // text node with whitespace only
    if (curr.type === 'TextNode' && /^[\s]*$/.test(curr.chars)) { continue; }

    // has multiple root elements if we've been here before
    if (nodeCount++ > 0) { return false; }

    if (curr.type === 'ComponentNode' || curr.type === 'ElementNode') {
      lastComponentNode = curr;
      lastIndex = i;
    }
  }

  if (!lastComponentNode) { return null; }

  if (lastComponentNode.type === 'ComponentNode') {
    let tag = lastComponentNode.tag;
    if (tag.charAt(0) !== '<') { return null; }
    return tag.slice(1, -1);
  }

  return null;
}


text:
Clóvis Ferro Costa (Passagem Franca,  – Rio de Janeiro, ) foi um advogado brasileiro.

Filho do juiz de direito José de Maria Costa e de Hilda de Maria Ferro Costa. Estudou na Faculdade de Direito da Universidade Federal do Pará, onde obteve o diploma de advogado em 1943.

Foi um dos fundadores da União Democrática Nacional (UDN) no Pará, sendo que nas eleições de 3 de outubro de 1950 elegeu-se deputado estadual por tal legenda. Assumindo sua cadeira em fevereiro do ano seguinte, na Assembléia Legislativa foi líder do governo de 1951 a 1954 e presidente da Comissão de Constituição e Justiça de 1951 a 1955. Foi reeleito no pleito de outubro de 1954, foi reconduzido à presidência dessa comissão, permanecendo nela até 1958.

Elegeu-se deputado federal em outubro de 1958, reelegendo-se quatro ano depois, sendo um dos poucos parlamentares da UDN que, durante o governo de João Goulart, defenderam as propostas de reformas originadas naquele governo. Exerceu seu mandato até 10 de abril de 1964, quando foi incluído na primeira lista de cassações publicada após o Ato Institucional nº 1, editado no dia anterior pela junta militar que depôs o presidente João Goulart. Desta forma, abandonou então a vida pública, recolhendo-se às atividades privadas.

Em setembro de 1989 foi nomeado consultor-geral da República na presidência José Sarney, em substituição a Saulo Ramos, nomeado ministro da Justiça. Permaneceu no cargo até o fim do governo Sarney, sucedido por Célio Silva.

Ver também
Eleições estaduais no Pará em 1958
Eleições estaduais no Pará em 1962
Lista de parlamentares cassados no Brasil
Epílogo de Campos
Comissão Provisória de Estudos Constitucionais
Lista de parlamentares do Pará

Advogados do Maranhão
Naturais de Passagem Franca
Ministros do Governo Sarney
Consultores-gerais da República do Brasil

redpajama_set_name: RedPajamaWikipedia


text:
A few clouds. Low 29F. Winds SW at 10 to 15 mph..
Jimmy Jackson
Nov 30, 2020 Updated Nov 30, 2020
James B. "Jimmy" Jackson, 52, of Madison, Indiana, died Tuesday, Nov. 24, 2020, at the Norton Hospital in Louisville.
He was born Dec. 18, 1967, in Madison, to Ronald Keith and Patricia A. Hall Jackson. He resided most of his life in Jefferson County and was a 1987 graduate of Southwestern High School. He was raised in the Chelsea Church of God, worked early in life for Royer Corporation and then for Grote Industries for 13 years. He married Deborah J. Elliott on Aug. 24, 1991, and they had a daughter. He was Baptized by his father at First Assembly of God Church in Madison. He worked for a time delivering for Domino's Pizza and Pizza Hut before being disabled by a stroke. He loved being with his daughter and watching television and listening to music. He liked to sing, loved sports, especially University of Kentucky basketball, Cincinnati Reds and New York Yankees baseball, the Louisville Cardinals and was an avid NASCAR fan. In his younger years, before he lost his eyesight, he loved fishing, exploring the woods, telling jokes and making people laugh. After loosing his sight he liked listening to audio books and music.
Survivors include his wife of 29 years; his daughter, Rebekah Faith Jackson of Madison; his son, Spenser Kyle Hanson of Indianapolis; his mother Patricia Hamm and her husband, Robert Hamm Jeffersonville, Indiana; his sisters, Theresa Joyce Nielsen of Jeffersonville, Lisa Michelle Kaelin of Durham, North Carolina and Stephanie Ann Via of Charlestown, Indiana; his caregiver of seven years, Kathy Bowen of Scottsburg, Indiana; and a granddaughter, several uncles and aunts, a great aunt, nieces and nephews, a great niece and great nephews, numerous cousins and other relatives.
He was preceded in death by his father and his grandparents, William Leonard and Sadie Hall and Thomas Glen and Thelma Lorraine Jackson.
Funeral services will be at 6 p.m. on Tuesday, Dec. 1, at the Morgan & Nay Funeral Centre in Madison. Interment will follow in the Springdale Cemetery in downtown Madison. Visitation is 4 p.m. until time of services Tuesday and then again after the funeral from 6:45 p.m. to 8 p.m. at the funeral home.
Memorial contributions may be made to the American Kidney Foundation or the American Diabetes Association. Cards are available at the funeral home. Online condolences, remembrances and memorials, www.morgan-nay.com.
1,346 have received COVID vaccine
State Champion
News of record
State Champ! Madison's Bilz upsets top-seed to win 106-pound title
Police investigating homicide near Holton

redpajama_set_name: RedPajamaCommonCrawl


Office - AMS field offices are listed in Appendix IV of the Catalog.
The information can be used by anyone as a basis for judging prevailing commodity values.
Market news information is disseminated by the news media, as well as by mimeographed reports, bulletin boards, telephone, facsimile machines, data networks, and telegraph.
Any public or private organization, or individual, including those located in the U.S.
Territories, may subscribe to existing market news reports or bulletins pertaining to specific agricultural commodities.
Anyone availing themselves of this service.
Requests for Market News should be made to the appropriate commodity division of the AMS field office (see Appendix IV of the Catalog for listing of office locations). The standard application forms as furnished by the Federal agency and required by OMB Circular No. A-102 must be used for this program.
(Salaries and expenses) FY 07 $32,182,000; FY 08 est $32,745,000; and FY 09 est not reported.
MB-40, "The Federal- State Market News Services;" MB-41, "Market News Service on Cotton;" MB-61, "The Market News Service on Fruits, Vegetables, Ornamental and Specialty Crops;" MB-44, "The Market News Service on Grain;" MB-63 "Market News Service on Dairy, MB-60 Market News Service and Poultry Products;" MB-43, "Market News Service for Tobacco and Naval Stores;" Annual Market Statistical Summaries are available for sale only through the AMS commodity divisions.


Chart showing changes of prices in standardized benchmark residential homes since 2005.
Lower Mainland real estate markets climbed modestly in the first two months of the year.
The Fraser Valley Real Estate Board (FVREB), which includes Surrey, White Rock and North Delta, reported a benchmark detached house price of $558,100 in February, up about 1.5 per cent from December and up 3.2 per cent year-over-year.
Townhouses were up 0.7 per cent to $296,700 from February of 2013 but the benchmark price of Fraser Valley apartments dropped 4.6 per cent from a year ago to $193,200.
The Real Estate Board of Greater Vancouver said its benchmark price for a typical detached house climbed 3.5 per cent from a year ago to $932,900.
Attached units were $458,300 – a 0.6 per cent one-year gain. Apartments were up 3.6 per cent over one year to $373,300.
The biggest one-year gains reported were for detached houses in Vancouver and South Burnaby, which are up more than seven per cent, while detached houses in North Delta and Langley were up six per cent.
The biggest recent drop was in Abbotsford apartments – their benchmark price is down 21 per cent from a year ago.
Other areas where prices have dropped include apartments in Squamish and Whistler – both down 13 per cent – and detached houses on Bowen Island and the Sunshine Coast, both down more than four per cent.
The most expensive market to buy a detached house remained the west side of Vancouver, where the benchmark price is $2.15 million, while the cheapest was Mission at $352,800.
It's taking less time for a home to sell – an average of 51 days in the Valley.
Both real estate boards reported sales are up significantly, reflecting a typical jump in buyer interest as spring approaches.


I am Kate: The Muddled Mum. This is my blog which documents how I have been making it through the muddle. This blog covers my experiences with miscarriage, mental health and all things motherhood.
I eagerly awaited motherhood; I longed for it. So imagine my surprise when pregnancy left me ill and isolated, and early motherhood left me in a fog of depression. I fought hard to get well again but struggled with my change in identity. I wasn't the mum I thought I would be, but I also wasn't the person I used to be. So again I fought to find my happiness – to find the new me. Eventually I was finally feeling like I knew what I was doing, so we decided it was time to give our daughter a sibling. But I miscarried. Twice (and then more times after that). After the challenges that had been thrown my way, this was the straw that broke the camels back. I found myself in a dark pit of depression and my chest was caged in anxiety. My blog was born in my time of need. My blog rescued me from a very dark place. It saw me through a mental health crisis. It helped me heal.
All of the opinions expressed on this blog are my own. Please do not take any of my advice to replace the advice of health care professionals. This is just an account of my experiences.
If you would like to work with me on a collaboration, product or experience review, or sponsorship, please get in touch using the form below or email contact@themuddledmum.com ! I would love to hear from you.
Disclosure: This blog contains affiliate links. You can read more about affiliate links and the kind you can find on this site in this post. I will let you know at the start of a post if it contains affiliate links (as of 7th Feb 2018) If I have received a product for free in order to review it, I will let you know explicitly in the post. If you would like to work with me, please do get in touch. Let's talk and see if we work well together. I'm into anything that gives me 'me' time, supports mental well being or makes me toddler happy.


text:
Eucynorta reducta is een hooiwagen uit de familie Cosmetidae.

Cosmetidae


2Mbps flat means, you'll get 2Mbps bandwidth all the time without FUP limitation. After FUP it BSNL will decrease the bandwidth as specified by the plan. Why is BSNL capping the 2Mbps connection to a Kbps download speed?. 2 Aug - 1 min - Uploaded by Lavian Cyders No Hack Is sriyoutube.com plan is of 2Mbps and my download speed reaches Kbps for unknown. Bandwidth (Download Speed) subject to technical feasibility, Upto 8 Mbps till 10 GB, Upto 1 Mbps beyond, Upto 8 Mbps till 15 GB, Upto 1 Mbps beyond, Upto 2.
BSNL Broadband minimum speed post FUP to 2Mbps. Download speed after the high-speed FUP quota increased from 1 Mbps to 2 Mbps. Bandwidth (Download Speed) subject to technical feasibility, Upto 10 Mbps till 30 GB, upto 2 Mbps beyond, Upto 10 Mbps till 60 GB, upto 2 Mbps beyond. 3 May 4 Mbps will be the initial download speed only for those broadband plans This plan offers GB broadband data at 2 Mbps till the daily data.


Available exclusively to Digitiser 2000 supporters, The Man's Daddy's Book of Popular Comedy Jokes contains almost 500 funnies - both classic (old) and new (new).
Additionally, there is a selection of features and whimsy from the legendary Digitiser character, on topics such as the history of popular comedy, and whether or not you're enjoying his book, as well as an introduction from "Blyan Adams".
If you'd like a copy of this luscious e-tome, all you have to do is sign up as one of our monthly Patreon donors, for just $10 (about £6.50), or pledge a one-off PayPal donation of £15.
Donors also get access to a bunch of other benefits, including exclusive videos, emails/messages, and more - and can sleep soundly in the knowledge that they're helping to keep this site alive.
Will the book be available in a paper or pig's skin format for us elderly people? I don't have an e-reader or a "hamtone 3000".
We're hoping to release a physical, holdable, special edition at some point before the end of the year, if there is sufficient demand (at which point our donors will be offered something new, to keep rewarding them). For the foreseeable, this PDF version - still readable on all phones and computers (hopefully) - is the only option, alas.
I can confirm the book is busting full of the funny.
Get in early. These first edition pdfs can only go up in value!


Archives|DR. BUTLER WARNS CITY ON FINANCES; Four-Year Agreement With the Banks Is Only a Palliative, Not a Cure, He Declares.
DR. BUTLER WARNS CITY ON FINANCES; Four-Year Agreement With the Banks Is Only a Palliative, Not a Cure, He Declares.


« Is Alexa Relevant Again?
Most blogs these days have ads, and unfortunately, most of us don't take the time to look at anyone else's ads because, well, we do have our own. However, I'm someone who does take a look at ads, because I want to see what other people might be marketing and how they might be doing it; learning how to be a good internet affiliate marketer is difficult work, after all.
It's occurred to me that basically there are two types of ads; product ads and advertiser ads. Product ads are, well, marketing products. For instance, if you look at my blog, you see that along the left side there I'm marketing products, starting with mine, then a couple of books. Some of my blog posts will have specific product ads, such as my last post (and I'm buying those suckers one day soon). Sometimes the product ad looks like an advertiser ad, but it's a product.
When we get to specific product items, such as computers, jewelry, or items like that, only myself and Sire, of Wassup Blog, seem to have actual items that we market, though his are more consistent than mine, as I have mine rotate on the side via TTZ Media, and within specific posts I decide on a product through Commission Junction and post it here.
Advertiser ads are promoting advertisers, and I've noticed that a lot of blogs have those, including mine. For instance, over on the left again, you see advertisements for Buy.com, National Geographic Store, Lapworks, etc, and in some posts, like this one, you see I end the post with an advertiser ad. Those take you to a page where you can look around and buy something.
But those aren't the only types of advertiser ads. Though I noticed that I don't have any right now, there are advertiser ads that market the advertisers themselves. For instance, on Caleb's blog, Market Secrets Blog, he has ads that market Adbrite and Pepperjam network, two that I haven't tried out yet. On Sire's blog, he has ads that market Onmarketer, Chitika, and ClixGalore. Yan's blog Thou Shall Blog, he's marketing Oiopublisher and Market Leverage. Even on Darren Rowse's blog, Problogger, even though all of his ads are actually sponsors, he's marketing Inlinks, LinkWorth, MLTV, and some others.
My question comes in wondering which types of ads for most people actually work for you. For the few of us who own up to our monthly blogging or online income, I don't see all that many of these ads working all that much for anyone. It's because of this that our friend John of 21st Century Affiliate Marketing has said that he's beginning to remove almost all of these types of ads to try something new, which he hopes will help stimulate online sales. Indeed, though I like having lots of different types of ads and products on my site, and other sites, it seems that these items rarely get all that many clicks, let alone sales, and yet I've noticed that I have clicked on some of the items listed above as advertiser ads, mainly because I'm thinking about marketing by using them to see how they fare. So, my impetus is purely as a test market, and not necessarily because I'm hoping to buy anything. But I believe that, when I actually decide to sign up via one of these links, if I got there through someone's site, they get some kind of monetary reimbursement from it.
Anyway, I'm wondering what y'all think of things like this? If our goal is really to make money with our blogs via affiliate marketing, it almost seems that the best way to do it is to write good content, try to jump up in some kind of rankings such as PR or Alexa, which seem to mean something to advertisers, and sell advertising on your site, ala Problogger. Visitors drive advertising, which is why I tried earlier this year to do my RSS subscriber contest, and goes back to a conversation I've had lately with our friend Mirjam of Me Myself And I Blog, as we've been talking about this issue of massive traffic that we both wrote about. Indeed, trying to obtain that kind of traffic just might be the only reason one may not care as much about targeted traffic.
So, share your thoughts and feelings on this topic, if you will. No, I didn't get into the contextual ads specifically, but I consider those as product ads, even if the products aren't always so clearly defined, because they're not trying to get you to go back to the site itself to help market them.
As to the other, yeah, one could go the other way, but then I couldn't have done the type of comparison I did if I'd gone your way. But private ads is something I think many of us shoot for in some fashion.
In my opinion, having an established blog, with great content and lots of visitors/subscribers etc etc, makes all the difference.. I remember hearing ShoeMoney (on track to make 3/4 million this year) say 75% of his income was from ads… but look how many visitors he has..
And it doesn't really mater WHAT product he offers, it'll sell because of his established blog..
I think early on is the toughest time to get advertisers to work with you.. but with patience and time, you can do very well!
Good point, Jake, but the truth in advertising is that some ads draw better than others, and even with Shoemoney's 75% income, I'm betting that one style works better than another for whatever reason.
I reckon product ads work if you use them in the right fashion. I've had more luck when I took the time to pick the right products for the right occasion, whereas now, due to time constraints, I just settle on one, plonk it in the post and hope for the best. Naturally this method isn't as productive.
I think you're correct, Sire, in looking for those right circumstances and selecting the proper product. It's kind of hard to do with most of what we write about, however; or so it seems to me.
I go with both types, but what Mitch calls 'Products Ads' sometimes brings in lifetime percentage of revenues from just one sale!
As Sire said, it also depends on the "occasion" which I interpret as meaning the particular economic cycle, holiday time, type of post, etc.
Well, this is the way way for attracting what Dennis calls 'Private ads'…it just depends on your particular blogging plans.
You're right, Caleb, people can bypass your links if they know how, which is one reason we try to build up a rapport with our visitors so that even if they know how to do it, they don't mind helping you out a little bit.
I think my content is what's gotten me my Alexa ranking, though it could be how much content I have rather than what I'm writing. Still, it is what it is; it can always be better, but at least it's moving in the right direction.
Well just for added encouragement, my site MSB just jumped from PR1 to PR2 from content writing and consistently going after backlnks and this was while I had almost a week of my site being out of commission due to tech glitches!
Good deal, Caleb, although remember, don't worry as much about PR.
thanks for the mention Mitch, only just saw it now because I´ve been a bit distracted this past week … I wonder if my boyfriend would make for a good secretary hahaha!
hahaha, haven´t touched the wine yet, but I have added the edit comment plugin, just for you Mitch ROFL plus it got me to have a look at a few others.
Didn´t get to the threaded comment plugin, still couldn´t get it to work even with the update, and I kinda don´t wanna go in to change the core of the theme …. so I added a "quote" plugin instead enabling you to respond to comments directly.
That´ll do for now, let´s open that bottle!
Well, as I was getting all of your test emails, I was wondering what the heck you were up to. lol Hey, we take what we can get; enjoy that bottle (I have cake, which my wife is presently hiding from me).


text:
In the month since Apple opened an online software clearinghouse called the App Store, users have downloaded more than 60 million programs for the iPhone, Chief Executive Steve Jobs said in an interview at Apple's headquarters. While most of those applications were free, Apple sold an average of $1 million a day in applications for a total of about $30 million in sales over the month, Mr. Jobs said.
IPhone Software Sales Take Off: Apple's Jobs – WSJ.com.
Categorized as Observations
Gmail down yesterday?


Here is a super easy and patriotic way to spruce up your home for the 4th of July with these pretty DIY Patriotic Candles.
DIY Patriotic Candles are easy to make and versatile since you can change the decorations to other themes. You can keep the base up all year and change out the center to match the current season. It's a great way to add simple beautiful decoration to your home!
They're also pretty cheap to make. I found the pillar candles on sale and only spent about $3 for the other pieces.
Start by hot gluing the burlap/lace ribbon onto each of the 3 pillar candles.
This will be the neutral base that can be used all year long – not just for the 4th of July.
For the bandanna strips, make a small cut in the bandanna about an inch apart and tear the cloth off.
These may be a little bit uneven and unraveled, but that is ok for this project. It makes it look more "Pioneer and Down-Home American." Perfect for a 4th of July celebration if you ask me.
Next you will need to make a red, white, and blue felt flower. For these you will need to cut 2 inch wide, and 4 inch long strips of each color of felt. Fold them over and hot glue them each together.
Then you will want to make 1/4 inch cuts, about halfway down – all along the strip of felt. These cuts make the "petals" of the felt flowers.
Now you simply roll up this felt strip into a flower. Roll it up just a bit and glue it, then continue rolling it up until you form your flower. Cover the bottom with more hot glue to make sure it is secure and stays together.
Now you need to match up a flower and bandanna of different colors. For example, the blue flower should go on the red bandana strip, the white flower should go on the multi-colored strip, and the red flower should go on the blue strip. Glue the flowers onto the middle of the strips and let them fully dry.
Now you have a red, white, and blue bandanna strip for each of the Patriotic candles. Tie the bandana in a bow around the center of the candle. No need to glue on this step.
And that's how you make DIY Patriotic Candles!
How festive. I love them. They came out awesome!
We don't typically do a lot of decorating for the 4th of July but I have a huge pinwheel that has a festive red, white & blue theme that I place in my front flower garden. We also put up US flags to honor our veterans and active military personnel.
This is such a cute idea. I love how easy it is to make, and it looks great!
What a great way to spruce up the house for the holiday. They look so easy to make.
These are adorable. It would be fun to make cuffs for every holiday and place them over my luminary candles. Thanks for a creative idea to celebrate the holidays.
Cuffs for each holiday – LOVE it!!
I love that you jazzed up boring candles with fabrics and lace to make some cute decorations!
Those are so cute and patriotic! Perfect for over our fireplace mantle this season!
How cute are these?!?! Perfect mantle decorations.
I absolutely love these! What a great craft for the kids.
These are fantastic! I especially love how they incorporate a patriotic red, while and blue theme without looking cheap or cheesy.
These turned out really nice! I love crafts such as this.
This is such a cute idea. I will have to make some up for the fourth of July. I love adding candles to a room.
These turned out fantastic. This is such a cute and simple project to do.
I love this idea. I could use this idea for a bunch of holidays. I am going to try this for sure.
These are gorgeous & I love that they could be changed for different holidays.
These are gorgeous and so simple to make- what a wonderful way to decorate a mantle for the season. Even better the candles can be re-purposed after the fourth!
How festive! I' definitely be making these soon, thanks!
So easy to do and gorgeous! My daughter loves crafts, this would be a great project for us both now for the summer!