Hazelcast SQL - SHOW DATA CONNECTIONS Örnekleri

Örnek

Şöyle yaparız

sql> SHOW DATA CONNECTIONS;
+--------------------+----------------------------------------+
|name                |types                                   |
+--------------------+----------------------------------------+
+--------------------+----------------------------------------+

Hazelcast DataConnectionService Arayüzü

Giriş

Şu satırı dahil ederiz 

import com.hazelcast.dataconnection.impl.DataConnectionService;

Kalıtım şöyle

DataConnectionService 

  InternalDataConnectionService

    DataConnectionServiceImpl

getAndRetainDataConnection metodu

Belirtilen isme sahip DataConnection nesnesini döndürür. DataConnection nesnesi Mapping ilk yaratıldığında oluşturulur

Örnek

Şöyle yaparız

DagBuildContext context = ...;
JdbcDataConnection dataConnection = context
  .getNodeEngine()
  .getDataConnectionService()
  .getAndRetainDataConnection(dataConnectionName, JdbcDataConnection.class);

Hazelcast SQL - JDBC İçin CREATE DATA CONNECTION Örnekleri

Giriş

jdbcUrl

user

password

parametreleri verilebilir

CREATE DATA CONNECTION IF NOT EXISTS

Örnek

Şöyle yaparız

CREATE DATA CONNECTION IF NOT EXISTS maria
TYPE Jdbc 
SHARED
OPTIONS (
  'jdbcUrl' = 'jdbc:mariadb://35.195.166.24:3306/transaction-monitor-payments',
  'user' = 'admin',
  'password' = 'password1' 
)

CREATE DATA CONNECTION 

Örnek

Şöyle yaparız. 

CREATE DATA CONNECTION mysql2
TYPE JDBC 
NOT SHARED 
OPTIONS('jdbcUrl'='jdbc:mysql://localhost:3306/mydb', 'user'='root', 'password'='123');

Örnek

Şöyle yaparız. 

CREATE DATA CONNECTION mysql2
TYPE JDBC 
SHARED 
OPTIONS('jdbcUrl'='jdbc:mysql://localhost:3306/mydb', 'user'='root', 'password'='123');

Hazelcast Jet SourceProcessors Sınıfı - ProcessorMetaSupplier Yaratır

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.core.processor.SourceProcessors;

Bu sınıf ProcessorMetaSupplier döndüren bir sürü metod sağlıyor.

readMapP metodu

İmzası şöyle

public static ProcessorMetaSupplier readMapP(String mapName);

İsmi belirtilen IMap nesnesini okumak içindir

streamMapP metodu

İmzası şöyle

public static <K, V> ProcessorMetaSupplier streamMapP(
  String mapName,
  JournalInitialPosition initialPos,
  EventTimePolicy<? super Entry<K, V>> eventTimePolicy
)

İsmi belirtilen IMap nesnesine yapılan değişklikleri yani Journal event'lerini okur

Hazelcast Jet AssertionSinks Sınıfı

Giriş

Şu satırı dahil ederiz 

import com.hazelcast.jet.pipeline.test.AssertionSinks;

Testlerde kullanmak için TestSources sınıfı var. Ama TestSinks diye bir şey yok. Sinks veya 

 AssertionSinks kullanılabilir

assertCollectedEventually metodu

Source nesneden gelen elemanları bir tane listede saklar. Elemanlar serialization işlemine maruz kalmazlar. Eğer bu test Sink nesnesi yerine bir başka Sink kullanmak istersek nesnenin Serializable olması gerekir. 

Örneğin Kafka'dan bir şey okumak istersek org.apache.kafka.connect.source.SourceRecord nesnesi Serializable olmadığı için Failed to serialize 'org.apache.kafka.connect.source.SourceRecord' hatası alırız. Kafka'dan gelen nesneyi bir projection metodu veya .map() çağrısı ile kendi nesnemize çevirmek gerekir

Örnek

Şöyle yaparız

Pipeline pipeline = Pipeline.create();
  ...
  .writeTo(Sinks.logger());
  .writeTo(AssertionSinks.assertCollectedEventually(60,
                        list -> assertEquals(100, list.size())));

clc komutu

Kurulum

https://github.com/hazelcast/hazelcast-commandline-client/releases adresinden bir sürüm indirilir

-c seçeneği

Örnek

Şöyle yaparız

clc -c sqltest sql "SELECT * FROM abc"

completion

Örnek

Desteklenen kabukları görmek için şöyle yaparız

clc completion --help

bash kabuğuyla ilgili yardım için şöyle yaparız

clc completion bash --help

map seçeneği

Örnek

Şöyle yaparız

clc map set -n mymap "mykey" "myvalue" --log.path stderr
clc map get -n mymap "mykey"--log.path stderr

GenericMapStore XML Konfigürasyonu

Giriş

XML konfigürasyonda sadece external-data-store-ref alanı mecburi. Gerisi isteğe bağlı

Örnek

Elimizde iki external data store olsun Burada isim olarak my-mysql-database kullanılıyor

<hazelcast>
  <external-data-store name="my-mysql-database">
    <class-name>com.hazelcast.datastore.JdbcDataStoreFactory</class-name>
    <properties>
      <property name="jdbcUrl">jdbc:mysql://mysql.example.org:3306</property>
      <property name="username">my_user</property>
      <property name="password">my_password</property>
    </properties>
    <shared>true</shared>
  </external-data-store>
  <external-data-store name="my-other-database">
    <class-name>com.hazelcast.datastore.JdbcDataStoreFactory</class-name>
    <properties>
      <property name="jdbcUrl">jdbc:h2:mem:my-other-database</property>
    </properties>
    <shared>true</shared>
  </external-data-store>
</hazelcast>

my-mysql-database bağlantısını kullanmak şöyle yaparız.

<hazelcast>
  <map name="Foo">
    <map-store enabled="true">
      <class-name>com.hazelcast.mapstore.GenericMapStore</class-name>
      <properties>
          <property name="external-data-store-ref">my-mysql-database</property>
      </properties>
    </map-store>
</hazelcast>

Bu aslında şuna denk gelir. Burada TYPE değerinin JDBC olduğuna dikkat. Ayrıca mapping ismine __map-store ön eki ekleniyor.

CREATE MAPPING "__map-store.Foo"
TYPE JDBC
OPTIONS (
 'externalDataStoreRef' = 'my-mysql-database'
)

Örnek - Sütun İsimleri Alanları

Şöyle yaparız

<map name=“Staff”>
  <map-store enabled=“true”>
    <class-name>com.hazelcast.mapstore.GenericMapStore</class-name>
    <properties>
      <property name=“external-data-store-ref”>my-mysql-database</property>
      <property name=“table-name”>employee</property>
      <property name=“id-column”>empId</property>
      <property name=“columns”>empName</property>
    </properties>
  </map-store>
 </map>

Bu sefer Hazelcast içinde yaratılan tablo şöyle

Örnek - id-column

Şöyle yaparız

<map name="shipstaticdata">
  <map-store enabled="true">
    <class-name>com.hazelcast.mapstore.GenericMapStore</class-name>
    <properties>
      <property name="external-data-store-ref">postgresdb</property>
      <property name="mapping-type">JDBC</property>
      <property name="id-column">name</property>
    </properties>
  </map-store>
</map>

<external-data-store name="postgresdb">
  <class-name>com.hazelcast.datastore.JdbcDataStoreFactory</class-name>
  <properties>
    <property name="jdbcUrl">jdbc:postgresql://localhost:5432/aisdata</property>
    <property name="username">postgres</property>
    <property name="password">NuZ8QxaXpjyxvlmL</property>
  </properties>
  <shared>true</shared>
</external-data-store>

HazelcastAPI MapStoreWrapper Sınıfı - Bir MapStore Türü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.map.impl.MapStoreWrapper;

İçinde bir tane MapLoader bir tane de MapStore nesnesi barındırır. Kod şöyle

public class MapStoreWrapper implements MapStore, MapLoaderLifecycleSupport {
  /**
   * An instance of {@link MapLoader} configured for this map
   * or {@code null} if none was provided.
   */
  private MapLoader mapLoader;
  /**
   * An instance of {@link MapStore} configured for this map
   * or {@code null} if none was provided.
   */
  private MapStore mapStore;

  private boolean withExpirationTime;

  private final String mapName;

  private final Object impl;
  ...
}

HazelcastAPI DataSerializerHook Arayüzü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.internal.serialization.DataSerializerHook;

Açıklaması şöyle

In Hazelcast, a DataSerializerHook is a mechanism to register custom serializers for objects that are not natively serializable. 

...

The DataSerializerHook is an interface that you can implement to register custom serializers for your objects. You can use this interface to define how your objects will be serialized and deserialized. When Hazelcast encounters an object that needs to be serialized or deserialized, it will check if a serializer is registered for that object. If there is a serializer, Hazelcast will use it to convert the object into a format that can be transmitted over the network. If there is no serializer, Hazelcast will throw an exception.

Bu arayüzden kalıtan bir sürü sınıf var. Bunlardan bir tanesi

JetDataSerializerHook

Hazelcast Jet HazelcastTableFunction Sınıfı

Kalıtım şöyle

org.apache.calcite.sql.SqlFunction

HazelcastTableFunction

  HazelcastTableSourceFunction

    HazelcastDynamicTableFunction

    HazelcastSpecificTableFunction

      SeriesGeneratorTableFunction : GENERATE_SERIES

      StreamGeneratorTableFunction : GENERATE_STREAM

  HazelcastWindowTableFunction

    HazelcastHopTableFunction

    HazelcastTumbleTableFunction

HazelcastAPI IMap.loadAll metodu

Giriş

İmzası şöyle. Asenkron olarak tüm birincil anahtar değerleri veya sadece belirtilen değerleri yükler.

void loadAll(boolean replaceExistingValues);
void loadAll(@Nonnull Set<K> keys, boolean replaceExistingValues);

Event Thread

Giriş

Listener gibi şeyleri tetikleyen thread bu. 

Örnek

EntryListener arayüzünün entryExpired() metodunu Event Thread tetikler

public void entryExpired(EntryEvent<String, Claim> event) {
   ...
}

Test Altyapısı - JdbcSqlTestSupport Sınıfı - TestContainers Kullanarak Veri Tabanı Yaratır

Giriş

Kalıtım şöyle

HazelcastTestSupport

  JetTestSupport

    SimpleTestInClusterSupport

      SqlTestSupport

        JdbcSqlTestSupport

          FooTest

initialize metodu

Bir tane TestDatabaseProvider nesnesi alır. TestContainers JDBC kullanarak bir test veri tabanı yaratır. Kod şöyle

public abstract class JdbcSqlTestSupport extends SqlTestSupport {
    
  public static void initialize(TestDatabaseProvider provider) {
    initialize(provider, smallInstanceConfig());
  }
  ...
}

Hazelcast SQL information_schema Schema

Giriş

Bu schema altında iki tane tablo var. Açıklaması şöyle

Currently, two tables are exposed:

mappings: contains information about existing mappings

columns: contains information about mapping columns

1. columns Tablosu

Sütunlar şöyle

1. column_name

2. data_type

3. is_nullable

Örnek

Şöyle yaparız

String query = "SELECT * FROM information_schema.columns WHERE table_name = ? 
  ORDER BY ordinal_position ASC";
try (SqlResult result = sqlService.execute(query, mapping)) {
  return StreamSupport
    .stream(result.spliterator(), false)
    .map(row -> {
      String name = row.getObject("column_name");
      SqlColumnType type = SqlColumnType.valueOf(row.getObject("data_type"));
      boolean isNullable = Boolean.parseBoolean(row.getObject("is_nullable"));
      return new SqlColumnMetadata(name, type, isNullable);
    })
    .collect(Collectors.toList());
}

2. mappings Tablosu

Sütunlar şöyle

1. table_catalog

Hep hazelcast

2. table_schema

Hep public

3. table_name

Mesela mykafka1

4. mapping_external_name

Mesela mykafka1

5. mapping_type

Mesela Kafka

6. mapping_options

CREATE MAPPING cümlesinde belirtilen OPTION alanlarını içerir.

Mesela

{
 "valueFormat":"json-flat",
 "bootstrap.servers":""a.b.c.d:9092",
 "security.protocol":"SASL_SSL",
 "sasl.jaas.config":"org.apache.kafka.common.security.plain.PlainLoginModule required username="..." password="...";",
"sasl.mechanism":"PLAIN",
"auto.offset.reset":"earliest"
}"

Örnek

Şöyle yaparız. Bunun yerine SHOW MAPPINGS de kullanılabilir

SELECT table_name, mapping_external_name, mapping_type FROM information_schema.mappings

HazelcastAPI PermissionConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.PermissionConfig;

Örnek

MessageTask şöyledir. Bu işlem için ACTION_SUBMIT iznini gerektirir.

public class JetUploadJobMetaDataTask extends
        AbstractJetMessageTask<JetUploadJobMetaDataCodec.RequestParameters, Boolean> {

  @Override
  public String[] actions() {
    return new String[]{ActionConstants.ACTION_SUBMIT};
  }
  ...
}

AbstractJetMessageTask şöyledir. Yani Job için izinleri döndürür.

abstract class AbstractJetMessageTask<P, R> extends AbstractInvocationMessageTask<P> {
   
  @Override
  public final Permission getRequiredPermission() {
    String[] actions = actions();
    if (actions != null) {
      return new JobPermission(actions);
    }
    return null;
  }
  ...
}

Member'ı çalıştırırken şöyle yaparız. addClientPermissionConfig() ile  dev istemcisinin ACTION_SUBMIT yeteneği etkin olur. Job submit edebilir. Çünkü JetUploadJobMetaDataTask nesnesi ACTION_SUBMIT iznine ihtiyaç duyar

Config config = ...;
SecurityConfig securityConfig = config.getSecurityConfig();
securityConfig.setEnabled(true);

String principal = "dev"
securityConfig.addClientPermissionConfig(
  new PermissionConfig(PermissionConfig.JOB, "", principal)
    .addAction(ActionConstants.ACTION_SUBMIT));

securityConfig.addClientPermissionConfig(
  new PermissionConfig(PermissionConfig.ALL, "", principal)
    .addAction(ActionConstants.ACTION_ALL));

Hazelcast Jet Traversers Sınıfı

Giriş
Şu satırı dahil ederiz 

import com.hazelcast.jet.Traversers;

Bu sınıf Traverser arayüzünü gerçekleştiren static factory metodlar sunar. Traverser arayüzü bir @FunctionalInterface O yüzden çoğu gerçekleştirim sadece lambda ile yapılabilir.

stream metodu

Kod şöyle

public static <T> Traverser<T> traverseStream(@Nonnull Stream<T> stream) {
  return traverseSpliterator(stream.spliterator()).onFirstNull(stream::close);
}

Stream nesnesini Spliterator nesnesi çevirir ve dolaşır. Altta kullanılan kod şöyle. Burada next() metodu Spliterator nesnesi tryAdvance() ile ilerlemişse sonraki nesneyi döner ve sonraki nesneyi finally block içinde tekrar null yapar. Böylece tryAdvance() false dönerse null döner ve stream::close da çalışır.

private static class SpliteratorTraverser<T> implements Traverser<T>, Consumer<T> {
  private final Spliterator<T> spliterator;
  private T nextItem;

  SpliteratorTraverser(Spliterator<T> spliterator) {
    this.spliterator = spliterator;
  }

  @Override
  public T next() {
    try {
      boolean advanced = spliterator.tryAdvance(this);
      if (advanced) {
        requireNonNull(nextItem);
      }
      return nextItem;
    } finally {
      nextItem = null; // Burada nesneyi null yap
    }
  }

  @Override
  public void accept(T t) {
    nextItem = t;
  }
}

traverseIterable metodu

İmzası şöyle

public static <T> Traverser<T> traverseIterable(Iterable<? extends T> iterable)

HazelcastAPI AddressPicker Arayüzü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.instance.AddressPicker;

Kalıtım şöyle

AddressPicker 

  AdvancedNetworkAddressPicker

  DefaultAddressPicker

  DelegatingAddressPicker

getPublicAddress metodu

Örnek

Şöyle yaparız

NetworkConfig networkConfig = config.getNetworkConfig();
networkConfig.setPublicAddress("localhost");
DefaultAddressPicker addressPicker = new DefaultAddressPicker(config, logger);
addressPicker.pickAddress();

// localhost:5701
Address actual = addressPicker.getPublicAddress(null);

DelegatingAddressPicker Sınıfı

getPublicAddressByPortSearch metodu

Bu metod ServerSocketHelper.tryOpenServerSocketChannel() çağrısı yaparak döngü içinde server socket olarak kullanılabilecek tüm portları dener

HazelcastAPI DistributedObject Arayüzü

Şu satırı dahil ederiz

import com.hazelcast.core.DistributedObject;

getName metodu

Örnek

Şöyle yaparız

HazelcastInstance hazelcastInstance = HazelcastClient.newHazelcastClient();

Collection<DistributedObject> distributedObjects = hazelcastInstance
  .getDistributedObjects();
for (DistributedObject object : distributedObjects) {
  if (object instanceof IMap) {
    IMap map = hazelcastInstance.getMap(object.getName());
    System.out.println("Mapname=" + map.getName());
    map.entrySet().forEach(System.out::println);
  }
}
hazelcastInstance.shutdown();

HazelcastAPI ITopic Arayüzü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.topic.ITopic;

Publish/Subscribe yöntemi sağlar. Yayınlanan mesajı tüm aboneler alırlar

constructor - Reliable Topic

Örnek

Şöyle yaparız

HazelcastInstance hz = Hazelcast.newHazelcastInstance();

ITopic<Long> topic = hz.getReliableTopic("sometopic");
long messageId = ...;

topic.publish(messageId);

addMessageListener metodu

Örnek

Publisher için şöyle yaparız

public class Trader {
  private HazelcastInstance hazelcast;
  private ITopic<TradeMessage> tradeTopic;
    
  public Trader() {
    hazelcast = Hazelcast.newHazelcastInstance();
    tradeTopic = hazelcast.getTopic("trades");
  }
    
  public void trade(String symbol, int quantity, double price) {
    TradeMessage trade = new TradeMessage(symbol, quantity, price);
    tradeTopic.publish(trade);
  }
}

Listener için şöyle yaparız. Listener hem member hem de client tarafında takılabilir.

public class TradeListener implements MessageListener<TradeMessage> {
  private HazelcastInstance hazelcast;
  private ITopic<TradeMessage> tradeTopic;
    
  public TradeListener() {
    hazelcast = Hazelcast.newHazelcastInstance();
    tradeTopic = hazelcast.getTopic("trades");
    tradeTopic.addMessageListener(this);
  }
    
  public void onMessage(Message<TradeMessage> message) {
    TradeMessage trade = message.getMessageObject();
  }
}

Takılan her bir MessageListener nenesi için MessageRunner sınıfından kalıtan

ReliableMessageRunner veya 

ClientReliableMessageRunner 

yaratılır

MessageRunner belirtilen ringBuffer nesnesinden okuma işlemi başlatır. İşlemin sonucunu kendisine dışarıdan verilen Executor üzerinde çalıştırır

publish metodu

Örnek

Şöyle yaparız

ITopic topic = ...;
topic.publish(objectToStore);

removeMessageListener

Örnek

Şöyle yaparız

HazelcastInstance hazelcastInstance = ...;


ITopic topic = hazelcastInstance.getReliableTopic("topic_name");

String listenerId = topic.addMessageListener(new TopicListener());
topic.removeMessageListener(listenerId);

Şöyle yaparız

class TopicListener implements MessageListener {

  @Override
  public void onMessage(Message message) {}

}