Работает ли Kafka для потоковой передачи на Ignite в транзакционном режиме ACID? - PullRequest
Новая
       21

Работает ли Kafka для потоковой передачи на Ignite в транзакционном режиме ACID?

0 голосов
/ 08 апреля 2019

У меня есть распределенная база данных в Apache Ignite и потоковая служба Apache Kafka, которая передает данные в кластер Ignite. Стример Кафки работает следующим образом

  1. Создание узла зажигания для поиска кластера
  2. Запустить Kafka Streamer Singleton в качестве службы в кластере
  3. Выключить узел зажигания

Кластер Ignite находится в транзакционном режиме, однако я не уверен, гарантирует ли это ACID или только включает его. Может ли этот потоковый сервис для Ignite считаться ACID?

Вот код для стримера Кафки: 

public class IgniteKafkaStreamerService implements Service {

private static final long serialVersionUID = 1L;

@IgniteInstanceResource
private Ignite ignite;
private KafkaStreamer<String, JSONObject> kafkaStreamer = new KafkaStreamer<>();
private IgniteLogger logger;

public static void main(String[] args) throws InterruptedException {
    TcpDiscoverySpi spi = new TcpDiscoverySpi();

    TcpDiscoveryMulticastIpFinder ipFinder = new TcpDiscoveryMulticastIpFinder();

    // Set Multicast group.
    //ipFinder.setMulticastGroup("228.10.10.157");

    // Set initial IP addresses.
    // Note that you can optionally specify a port or a port range.
    ipFinder.setAddresses(Arrays.asList("127.0.0.1:47500..47509"));

    spi.setIpFinder(ipFinder);

    IgniteConfiguration cfg = new IgniteConfiguration();

    // Override default discovery SPI.
    cfg.setDiscoverySpi(spi);
    Ignite ignite = Ignition.getOrStart(cfg);

    // Deploy data streamer service on the server nodes.
    ClusterGroup forServers = ignite.cluster().forServers();
    IgniteKafkaStreamerService streamer = new IgniteKafkaStreamerService();
    ignite.services(forServers).deployClusterSingleton("KafkaService", streamer);
    ignite.close();
}


@Override
public void init(ServiceContext ctx) {
    logger = ignite.log();
    IgniteDataStreamer<String, JSONObject> stmr = ignite.dataStreamer("my_cache");
    stmr.allowOverwrite(true);
    stmr.autoFlushFrequency(1000);
    List<String> topics = new ArrayList<>();
    topics.add(0,"IoTData");

    kafkaStreamer.setIgnite(ignite);
    kafkaStreamer.setStreamer(stmr);
    kafkaStreamer.setThreads(4);
    kafkaStreamer.setTopic(topics);
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "NiFi-consumer");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.242:9092");
    props.put("key.deserializer", StringDeserializer.class.getName());
    props.put("value.deserializer", StringDeserializer.class.getName());
    props.put("group.id", "hello");
    kafkaStreamer.setConsumerConfig(props);
    kafkaStreamer.setSingleTupleExtractor(msg -> {
        JSONObject jsonObj = new JSONObject(msg.value().toString());
        String key = jsonObj.getString("id") + "," + new Date(msg.timestamp());
        JSONObject value = jsonObj.accumulate("date", new Date(msg.timestamp()));

        return new AbstractMap.SimpleEntry<>(key, value);

    });
}

@Override
public void execute(ServiceContext ctx) {
    kafkaStreamer.start();
    logger.info("KafkaStreamer started.");
}

@Override
public void cancel(ServiceContext ctx) {
    kafkaStreamer.stop();
    logger.info("KafkaStreamer stopped.");
}

}

1 Ответ

2 голосов
/ 08 апреля 2019

KafkaStreamer использует IgniteDataStreamer реализации под капотом.IgniteDataStreamer не является транзакционным по своей природе, поэтому нет никаких транзакционных гарантий.

...