package main
import (
	"syscall"
	"unsafe"
)
func a() int { return 1 }
func b() int { return 2 }
func getPage(p uintptr) []byte {
	return (*(*[0xFFFFFF]byte)(unsafe.Pointer(p & ^uintptr(syscall.Getpagesize()-1))))[:syscall.Getpagesize()]
}
func rawMemoryAccess(b uintptr) []byte {
	return (*(*[0xFF]byte)(unsafe.Pointer(b)))[:]
}
func assembleJump(f func() int) []byte {
	funcVal := *(*uintptr)(unsafe.Pointer(&f))
	return []byte{
		0x48, 0xC7, 0xC2,
		byte(funcVal >> 0),
		byte(funcVal >> 8),
		byte(funcVal >> 16),
		byte(funcVal >> 24), // MOV rdx, funcVal
		0xFF, 0x22,          // JMP rdx
	}
}
func replace(orig, replacement func() int) {
	bytes := assembleJump(replacement)
	functionLocation := **(**uintptr)(unsafe.Pointer(&orig))
	window := rawMemoryAccess(functionLocation)
	
	page := getPage(functionLocation)
	syscall.Mprotect(page, syscall.PROT_READ|syscall.PROT_WRITE|syscall.PROT_EXEC)
	
	copy(window, bytes)
}
func main() {
	replace(a, b)
	print(a())
}

fetchSizeBytes  = 1024 * 1024 * 2 // 1048576=1M by default FetchRequest
fetchMaxWait    = 10000           // by default
forceFromStart  = false

class PartitionManager {
    SortedSet<Long> _pending, failed = new TreeSet();
    LinkedList<MessageAndRealOffset> _waitingToEmit = new LinkedList();
    func EmitState next(SpoutOutputCollector collector) {
        if (this._waitingToEmit.isEmpty()) {
            // 如果内存里数据都发出，就调用kafka consumer一次性批量填充内存_waitingToEmit
            // 填充时，如果发现failed里有东西，那么就从head of failed(offset) FetchRequest: 重发机制
            this.fill();
        }
        // 从LinkedList _waitingToEmit里取一条消息
        MessageAndRealOffset toEmit = (MessageAndRealOffset)this._waitingToEmit.pollFirst();
        // emit时指定了messageID
        // BasicBoltExecutor.execute会通过template method自动执行_collector.getOutputter().ack(input)
        // 即KafkaSpout.ack -> PartitionManager.ack
        collector.emit(tup, new PartitionManager.KafkaMessageId(this._partition, toEmit.offset));
        // 该tuple处于pending state
    }
    // Note: a tuple will be acked or failed by the exact same Spout task that created it
    func ack(Long offset) {
        this._pending.remove(offset)
    }
    func fail(Long offset) {
        this.failed.add(offset);
        // kafka consumer会reset offset to the failed msg，重新消费
    }
}
class TopologyBuilder {
    public BoltDeclarer setBolt(String id, IBasicBolt bolt, Number parallelism_hint) {
        return setBolt(id, new BasicBoltExecutor(bolt), parallelism_hint);
    }
}
class BasicBoltExecutor {
    public void execute(Tuple input) {
         _collector.setContext(input);
         try {
             _bolt.execute(input, _collector);
             _collector.getOutputter().ack(input);
         } catch(FailedException e) {
             _collector.getOutputter().fail(input);
         }
    }
}

class MyBolt {
    public void execute(Tuple tuple) {
        _collector.emit(new Values(foo, bar))
        _collector.ack(tuple)
    }
}

Key： tablePrefix_rowPrefix_tableID_rowID
Value: [col1, col2, col3, col4]

Key: tablePrefix_idxPrefix_tableID_indexID_indexColumnsValue
Value: rowID

Key: tablePrefix_idxPrefix_tableID_indexID_ColumnsValue_rowID
Value: nil

var (
    tablePrefix     = []byte{'t'}
    recordPrefixSep = []byte("_r")
    indexPrefixSep  = []byte("_i")
)

    RandRead    RandWrite   SeqRead    SeqWrite
SSD 50          200         2300       1300
HDD 0.6         1.0         200        130

class KafkaController {
    partitionStateMachine PartitionStateMachine
    replicaStateMachine ReplicaStateMachine
    // controller选举，并LeaderChangeListener
    controllerElector ZookeeperLeaderElector 
}

class ControllerContext {
    epoch, epochZkVersion, correlationId
    controllerChannelManager // 维护controller与每个broker之间的socket conn
    // 初始化时从/brokers/ids里取得的，并在BrokerChangeListener里修改
    liveBrokersUnderlying: Set[Broker]
    // 初始化时从/brokers/topics里取得的，并在TopicChangeListener里修改
    allTopics: Set[String]
    // 初始化时从/brokers/topics/$topic里一一取得，并在如下情况下修改
    // - assignReplicasToPartitions.assignReplicasToPartitions
    // - updateAssignedReplicasForPartition
    // - TopicChangeListener
    // - ReplicaStateMachine.handleStateChange
    partitionReplicaAssignment: mutable.Map[TopicAndPartition, Seq[Int]] // AR
    // 初始化时从/brokers/topics/$topic/$partitionID/state一一取得，如下情况下修改
    // - PartitionStateMachine.initializeLeaderAndIsrForPartition
    // - PartitionStateMachine.electLeaderForPartition
    partitionLeadershipInfo: mutable.Map[TopicAndPartition, LeaderIsrAndControllerEpoch]
}

sendUpdateMetadataRequest(newBrokers)
// 告诉它它上面的所有partitions
replicaStateMachine.handleStateChanges(allReplicasOnNewBrokers, OnlineReplica)
// 让所有的NewPartition/OfflinePartition进行leader election
partitionStateMachine.triggerOnlinePartitionStateChange()

Replica {
    broker_id int
    partition Partition // a Replica belongs to a Partition
    log Log
    hw, leo long
    isLeader bool
}
Partition {
    topic string
    partition_id int
    leader Replica
    ISR Set[Replica]
    AR  Set[Replica] // assigned replicas
    zkVersion long   // for CAS
}

class PartitionStateMachine {
    // - NonExistentPartition
    // - NewPartition
    // - OnlinePartition
    // - OfflinePartition
    partitionState: mutable.Map[TopicAndPartition, PartitionState] = mutable.Map.empty
    topicChangeListener   // childChanges("/brokers/topics")
    addPartitionsListener // all.dataChanges("/brokers/topics/$topic")
    deleteTopicsListener  // childChanges("/admin/delete_topics")
}

class ReplicaStateMachine {
    replicaState: mutable.Map[PartitionAndReplica, ReplicaState] = mutable.Map.empty
    brokerRequestBatch = new ControllerBrokerRequestBatch
    // 只有controller关心每个broker的存活，broker自己是不关心的
    // 而且broker die只有2种标准，不会因为conn broken认为broker死
    // 1. broker id ephemeral znode deleted
    // 2. broker controlled shutdown
    brokerChangeListener  // childChanges("/brokers/ids")
}

{"jmx_port":-1,"timestamp":"1460677527837","host":"10.1.1.1","version":1,"port":9002}

partition0 broker(A, B, C)，A是leader
A GC很久，crontroller认为A死，让B成为leader，写zk ISR znode，正在此时A活了但还没有收到controller发来的RPC，此时A、B都是leader
如果client1连接A，clientB连接B，他们都发消息?
A活过来的时候，A认为它的ISR=(A,B,C)，clientA发的消息commit条件是A,B,C都ack by fetch request才可以，但B在promoted to leader后会先stop fetching from previous leader。
因此，A只有shrink ISR后才可能commit消息，但shrink时它会写zk，通过CAS失败，此时A意识到它已经不是leader

broker1，是controller
  在sessionTimeout*2/3=4s内还没有收到response，就会try next zk server
  zk connected，发现session(0x35c1876461ec9d3)已经expire了
  触发KafkaController.SessionExpirationListener {
      onControllerResignation()
      brokerState设置为RunningAsBroker // 之前是RunningAsController
      controllerElector.elect {
          switch write("/controller") {
              case ok:
                  onBecomingLeader {
                      // 1 successfully elected as leader
                      // Broker 1 starting become controller state transition
                      controllerEpoch.increment()
                      register listeners
                      replicaStateMachine.startup()
                      partitionStateMachine.startup()
                      brokerState = RunningAsController
                      sendUpdateMetadataRequest(to all live brokers)
                  }
              case ZkNodeExistsException:
                  read("/controller") and get leaderID
          }
      }
  }
broker0:
  ZookeeperLeaderElector.LeaderChangeListener被触发，它watch("/controller") {
      elect
  }