Flink与Spark Streaming在与kafka结合的区别！

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.getConfig.disableSysoutLogging
env.getConfig.setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000))
// create a checkpoint every 5 seconds
env.enableCheckpointing(5000)
// make parameters available in the web interface
env.getConfig.setGlobalJobParameters(params)

// create a Kafka streaming source consumer for Kafka 0.10.x
val kafkaConsumer = new FlinkKafkaConsumer010(
  params.getRequired("input-topic"),
  new SimpleStringSchema,
  params.getProperties)

val messageStream = env
  .addSource(kafkaConsumer)
  .map(in => prefix + in)

// create a Kafka producer for Kafka 0.10.x
val kafkaProducer = new FlinkKafkaProducer010(
  params.getRequired("output-topic"),
  new SimpleStringSchema,
  params.getProperties)

// write data into Kafka
messageStream.addSink(kafkaProducer)

env.execute("Kafka 0.10 Example")

this.kafkaFetcher = createFetcher(
      sourceContext,
      subscribedPartitionsToStartOffsets,
      periodicWatermarkAssigner,
      punctuatedWatermarkAssigner,
      (StreamingRuntimeContext) getRuntimeContext(),
      offsetCommitMode,
      getRuntimeContext().getMetricGroup().addGroup(KAFKA_CONSUMER_METRICS_GROUP),
      useMetrics);

final AtomicReference<Exception> discoveryLoopErrorRef = new AtomicReference<>();
this.discoveryLoopThread = new Thread(new Runnable() {
   @Override
   public void run() {
      try {
         // --------------------- partition discovery loop ---------------------

         List<KafkaTopicPartition> discoveredPartitions;

         // throughout the loop, we always eagerly check if we are still running before
         // performing the next operation, so that we can escape the loop as soon as possible

         while (running) {
            if (LOG.isDebugEnabled()) {
               LOG.debug("Consumer subtask {} is trying to discover new partitions ...", getRuntimeContext().getIndexOfThisSubtask());
            }

            try {
               discoveredPartitions = partitionDiscoverer.discoverPartitions();
            } catch (AbstractPartitionDiscoverer.WakeupException | AbstractPartitionDiscoverer.ClosedException e) {
               // the partition discoverer may have been closed or woken up before or during the discovery;
               // this would only happen if the consumer was canceled; simply escape the loop
               break;
            }

            // no need to add the discovered partitions if we were closed during the meantime
            if (running && !discoveredPartitions.isEmpty()) {
               kafkaFetcher.addDiscoveredPartitions(discoveredPartitions);
            }

            // do not waste any time sleeping if we're not running anymore
            if (running && discoveryIntervalMillis != 0) {
               try {
                  Thread.sleep(discoveryIntervalMillis);
               } catch (InterruptedException iex) {
                  // may be interrupted if the consumer was canceled midway; simply escape the loop
                  break;
               }
            }
         }
      } catch (Exception e) {
         discoveryLoopErrorRef.set(e);
      } finally {
         // calling cancel will also let the fetcher loop escape
         // (if not running, cancel() was already called)
         if (running) {
            cancel();
         }
      }
   }
}, "Kafka Partition Discovery for " + getRuntimeContext().getTaskNameWithSubtasks());

discoveryLoopThread.start();
kafkaFetcher.runFetchLoop();

// --------------------------------------------------------------------

// make sure that the partition discoverer is properly closed
partitionDiscoverer.close();
discoveryLoopThread.join();

// kick off the actual Kafka consumer
consumerThread.start();

this.consumerThread = new KafkaConsumerThread(
      LOG,
      handover,
      kafkaProperties,
      unassignedPartitionsQueue,
      createCallBridge(),
      getFetcherName() + " for " + taskNameWithSubtasks,
      pollTimeout,
      useMetrics,
      consumerMetricGroup,
      subtaskMetricGroup);

// this is the means to talk to FlinkKafkaConsumer's main thread
final Handover handover = this.handover;

try {
   this.consumer = getConsumer(kafkaProperties);
}

try {
   if (hasAssignedPartitions) {
      newPartitions = unassignedPartitionsQueue.pollBatch();
   }
   else {
      // if no assigned partitions block until we get at least one
      // instead of hot spinning this loop. We rely on a fact that
      // unassignedPartitionsQueue will be closed on a shutdown, so
      // we don't block indefinitely
      newPartitions = unassignedPartitionsQueue.getBatchBlocking();
   }
   if (newPartitions != null) {
      reassignPartitions(newPartitions);
   }

// get the next batch of records, unless we did not manage to hand the old batch over
if (records == null) {
   try {
      records = consumer.poll(pollTimeout);
   }
   catch (WakeupException we) {
      continue;
   }
}

try {
   handover.produce(records);
   records = null;
}

while (running) {
   // this blocks until we get the next records
   // it automatically re-throws exceptions encountered in the consumer thread
   final ConsumerRecords<byte[], byte[]> records = handover.pollNext();

// get the records for each topic partition
for (KafkaTopicPartitionState<TopicPartition> partition : subscribedPartitionStates()) {

   List<ConsumerRecord<byte[], byte[]>> partitionRecords =
         records.records(partition.getKafkaPartitionHandle());

   for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {
      final T value = deserializer.deserialize(
            record.key(), record.value(),
            record.topic(), record.partition(), record.offset());

      if (deserializer.isEndOfStream(value)) {
         // end of stream signaled
         running = false;
         break;
      }

      // emit the actual record. this also updates offset state atomically
      // and deals with timestamps and watermark generation
      emitRecord(value, partition, record.offset(), record);
   }
}

this.handover = new Handover();