statestore-subscriber.h

Go to the documentation of this file.

// Copyright 2012 Cloudera Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.


#ifndef STATESTORE_STATESTORE_SUBSCRIBER_H
#define STATESTORE_STATESTORE_SUBSCRIBER_H

#include <string>

#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>

#include "statestore/statestore.h"
#include "util/stopwatch.h"
#include "rpc/thrift-util.h"
#include "rpc/thrift-client.h"
#include "util/thread.h"
#include "util/metrics.h"
#include "gen-cpp/StatestoreService.h"
#include "gen-cpp/StatestoreSubscriber.h"

namespace impala {

class TimeoutFailureDetector;
class Status;
class TNetworkAddress;
class ThriftServer;

typedef ClientCache<StatestoreServiceClient> StatestoreClientCache;

//
//
//
class StatestoreSubscriber {
 public:
  //
  //
  StatestoreSubscriber(const std::string& subscriber_id,
      const TNetworkAddress& heartbeat_address,
      const TNetworkAddress& statestore_address,
      MetricGroup* metrics);

  typedef std::map<Statestore::TopicId, TTopicDelta> TopicDeltaMap;

  //
  //
  //
  typedef boost::function<void (const TopicDeltaMap& state,
                                std::vector<TTopicDelta>* topic_updates)> UpdateCallback;

  //
  Status AddTopic(const Statestore::TopicId& topic_id, bool is_transient,
      const UpdateCallback& callback);

  //
  Status Start();

  const std::string& id() const { return subscriber_id_; }

 private:
  const std::string subscriber_id_;

  TNetworkAddress heartbeat_address_;

  TNetworkAddress statestore_address_;

  boost::shared_ptr<StatestoreSubscriberIf> thrift_iface_;

  boost::shared_ptr<ThriftServer> heartbeat_server_;

  boost::scoped_ptr<impala::TimeoutFailureDetector> failure_detector_;

  boost::scoped_ptr<Thread> recovery_mode_thread_;

  boost::mutex lock_;

  bool is_registered_;

  boost::mutex registration_id_lock_;

  TUniqueId registration_id_;

  struct Callbacks {
    StatsMetric<double>* processing_time_metric;

    std::vector<UpdateCallback> callbacks;
  };

  typedef boost::unordered_map<Statestore::TopicId, Callbacks> UpdateCallbacks;

  UpdateCallbacks update_callbacks_;

  std::map<Statestore::TopicId, bool> topic_registrations_;

  typedef boost::unordered_map<Statestore::TopicId, int64_t> TopicVersionMap;
  TopicVersionMap current_topic_versions_;

  boost::scoped_ptr<StatestoreClientCache> client_cache_;

  MetricGroup* metrics_;

  BooleanProperty* connected_to_statestore_metric_;

  DoubleGauge* last_recovery_duration_metric_;

  StringProperty* last_recovery_time_metric_;

  StatsMetric<double>* topic_update_interval_metric_;

  MonotonicStopWatch topic_update_interval_timer_;

  StatsMetric<double>* topic_update_duration_metric_;

  MonotonicStopWatch heartbeat_interval_timer_;

  StatsMetric<double>* heartbeat_interval_metric_;

  StringProperty* registration_id_metric_;

  friend class StatestoreSubscriberThriftIf;

  //
  Status UpdateState(const TopicDeltaMap& incoming_topic_deltas,
      const TUniqueId& registration_id,
      std::vector<TTopicDelta>* subscriber_topic_updates, bool* skipped);

  void Heartbeat(const TUniqueId& registration_id);

  //
  void RecoveryModeChecker();

  Status Register();

  Status CheckRegistrationId(const TUniqueId& registration_id);
};

}

#endif