Hazelcast Jet JdbcDataConnection Sınıfı

Giriş

Şu satırı dahil ederiz 

import com.hazelcast.dataconnection.impl.JdbcDataConnection;

Kalıtım şöyle

DataConnection

  DataConnectionBase

    JdbcDataConnection

Bu sınıf direkt yaratılmıyor. Service örüntüsü kullanılarak dinamik olarak bulunuyor ve yaratılıyor. Bunu başaran sınıf ise JdbcDataConnectionRegistration sınıfı

getConnection metodu

Hikari'den alınmış veya Hikari kullanılmıyorsa yeni yaratılmış bir Connection nesnesi döndürür

listResources metodu

İmzası şöyle. . DataConnectionResource listesi döndürür. 

Collection<DataConnectionResource> listResources();

Veri tabanını keşfetmek içindir. Döndürülen listedeki tablolara DataConnection yaratılabilir. 

Ne Döndürür

JDBC açısından şu sonucu döndürür. 

connection.getMetaData().getTables();

Hazelcast Jet HazelcastPhysicalScan Arayüzü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.sql.impl.HazelcastPhysicalScan;

Eğer iki tablo birleştiriliyorsa ve sağdaki üzerinde indeks yoksa sanırım FullScanPhysicalRel kullanılır. Bu da nested loop gibi düşünülebilir

Kalıtım şöyle

org.apache.calcite.rel.RelNode

  PhysicalRel

    HazelcastPhysicalScan

      FullScanPhysicalRel : Bu sınıfı ilk olarak burada gördüm

      IndexScanMapPhysicalRel

Hazelcast SQL IMap + IMap JOIN

Örnek

Şöyle yaparız. Burada iki tane IMap join'leniyor

CREATE MAPPING client 
TYPE IMap 
OPTIONS (
  'keyFormat' = 'java', 
  'keyJavaClass' = 'java.lang.Long', 
  'valueFormat' = 'java', 
  'valueJavaClass' = 'com.foo.Client' 
)

CREATE MAPPING client2system 
TYPE IMap 
OPTIONS (
  'keyFormat' = 'java', 
  'keyJavaClass' = 'java.lang.Long', 
  'valueFormat' = 'java', 
  'valueJavaClass' = 'com.foo.Client2System' 
)

SELECT c.clientKey 
FROM client c 
JOIN client2system c2s on c2s.clientId = c.clientId
WHERE c2s.systemId = ? 
AND c2s.clientCode = ?

Hazelcast SQL Kafka + IMap JOIN

Örnek

Kafka için mapping yaratırız.

CREATE OR REPLACE MAPPING pizzastream(
  timestamp_ TIMESTAMP,
  pizza VARCHAR,
  user_id VARCHAR,
  quantity DOUBLE
)
TYPE Kafka
OPTIONS (
  'keyFormat' = 'varchar',
  'valueFormat' = 'json-flat',
  'auto.offset.reset' = 'earliest',
  'bootstrap.servers' = 'localhost:19092'
);

IMap için mapping yaratırız.

CREATE or REPLACE MAPPING recommender (
  __key BIGINT,
  user_id VARCHAR,
  extra1 VARCHAR,
  extra2 VARCHAR,
  extra3 VARCHAR )
TYPE IMap
OPTIONS (
  'keyFormat'='bigint',
  'valueFormat'='json-flat'
);

IMap'e veri yazarız

INSERT INTO recommender VALUES
(1, 'user_1', 'Soup','Onion_rings','Coleslaw'),
(2, 'user_2', 'Salad', 'Coleslaw', 'Soup'),
(3, 'user_3', 'Zucchini_fries','Salad', 'Coleslaw'),
(4, 'user_4', 'Onion_rings','Soup', 'Jalapeno_poppers'),
(5, 'user_5', 'Zucchini_fries', 'Salad', 'Coleslaw'),
(6, 'user_6', 'Soup', 'Zucchini_fries', 'Coleslaw'),
(7, 'user_7', 'Onion_rings', 'Soup', 'Jalapeno_poppers'),
(8, 'user_8', 'Jalapeno_poppers', 'Coleslaw', 'Zucchini_fries'),
(9, 'user_9', 'Onion_rings','Jalapeno_poppers','Soup');

Join işlemi yaparız

SELECT
    pizzastream.user_id AS user_id,
    recommender.extra1 as extra1,
    recommender.extra2 as extra2,
    recommender.extra3 as extra3,
     pizzastream.pizza AS pizza
FROM pizzastream
JOIN recommender
ON recommender.user_id = recommender.user_id 
AND recommender.extra2 = 'Soup';

Join sonucunu tekrar Kafka'ya göndeririz.

CREATE OR REPLACE MAPPING recommender_pizzastream(
  timestamp_ TIMESTAMP,
  user_id VARCHAR,
  extra1 VARCHAR,
  extra2 VARCHAR,
  extra3 VARCHAR,
  pizza VARCHAR
)
TYPE Kafka
OPTIONS (
  'keyFormat' = 'int',
  'valueFormat' = 'json-flat',
  'auto.offset.rest' = 'earliest',
  'bootstrap.servers' = 'localhost:19092'
);

CREATE JOB recommender_job AS SINK INTO recommender_pizzastream SELECT
    pizzastream.timestamp_ as timestamp_,
    pizzastream.user_id AS user_id,
    recommender.extra1 as extra1,
    recommender.extra2 as extra2,
    recommender.extra3 as extra3,
    pizzastream.pizza AS pizza
FROM pizzastream
JOIN recommender
ON recommender.user_id = recommender.user_id 
AND recommender.extra2 = 'Soup';

HazelcastAPI Kubernetes KubernetesClient Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.kubernetes.KubernetesClient;

Kubernetes API'ye bağlanarak REST çağrıları yapar

HazelcastAPI RingBufferConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.RingBufferConfig;

setCapacity metodu

Örnek

Şöyle yaparız

HazelcastInstance client = HazelcastClient.newHazelcastClient();
Config config = client.getConfig();
config.addRingBufferConfig(new RingbufferConfig("foo").setCapacity(12345));

Ringbuffer<Object> ringbuffer = client.getRingbuffer("foo");

HazelcastAPI MemberDomConfigProcessor Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.internal.config.MemberDomConfigProcessor;

Ortam değişkeni gibi konfigürasyonu kullanarak Config nesnesindeki alanları override eder

buildConfig metodu

Örnek bir stack trace şöyle. Burada HZ_NETWORK_JOIN_TCPIP_MEMBERS ortam değişkenine değer atadım

handleTcpIp:1659, MemberDomConfigProcessor (com.hazelcast.internal.config)

handleJoin:1485, MemberDomConfigProcessor (com.hazelcast.internal.config)

handleNetwork:963, MemberDomConfigProcessor (com.hazelcast.internal.config)

handleNode:302, MemberDomConfigProcessor (com.hazelcast.internal.config)

buildConfig:285, MemberDomConfigProcessor (com.hazelcast.internal.config)

lambda$overwriteMemberConfig$0:63, ExternalConfigurationOverride (com.hazelcast.internal.config.override)

apply:-1, 1435781140 (com.hazelcast.internal.config.override.ExternalConfigurationOverride$$Lambda$122)

overwrite:94, ExternalConfigurationOverride (com.hazelcast.internal.config.override)

overwriteMemberConfig:60, ExternalConfigurationOverride (com.hazelcast.internal.config.override)

hazelcast.xml - Set Ayarları

item-listeners

Örnek

Şöyle yaparız

<hazelcast>
    ...
    <set>
        <item-listeners>
            <item-listener include-value="true">
                com.yourpackage.ExampleItemListener
            </item-listener>
        </item-listeners>
    </set>
    ...
</hazelcast>

Compact Serialization CompactSerializableRegistration Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.internal.serialization.impl.compact.CompactSerializableRegistration;

Kod şöyle. Bir sınıf için tanımlı CompactSerializer nesnesini taşır

public class CompactSerializableRegistration {
  private final Class clazz;
  private final CompactSerializer compactSerializer;
  private final String typeName;
  ...
}

HazelcastAPI ExternalConfigurationOverride Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.internal.config.override.ExternalConfigurationOverride;

 Overriding Configuration sayfasındaki şeyler yapan sınıf bu. 

Ortam değişkenleri veya system properties ile atanan yeni değerleri Config veya ClientConfig nesnesine atar. Böylece dışarıdan verilen değişkenler kullanılabilir. Değişken isimlerinde bir örüntü izlenmiş. 

Bu sınıf bir DomConfigProcessor nesnesini çağırır

DomConfigProcessor

  AbstractDomConfigProcessor

    ClientDomConfigProcessor

    ClientFailoverDomConfigProcessor

    MemberDomConfigProcessor

Configuration Override Örnekleri

Örnek - HZ_NETWORK_PUBLICADDRESS

yaml dosyasındaki şu ayara denk gelir. NetworkConfig sınıfındaki publicAddress alanına değer atar.

hazelcast:
  network:
    public-address: 11.22.33.44:5555

Örnek - HZ_NETWORK_PUBLICADDRESS

Şöyle yaparız. NetworkConfig sınıfındaki publicAddress alanına değer atar.

Burada expose edilen port ile HZ_NETWORK_PUBLICADDRESS değişkeninde kullanılan port aynı olmalı. Ayrıca HZ_NETWORK_PUBLICADDRESS değişkenine 127.0.0.1 verilebilir. Bu örnekteki gibi 172.22.41.210 şeklindeki Host IP adresini kullanmaya gerek yok

services: 
  hazelcast: 
    container_name: hazelcast
    environment: 
      HZ_NETWORK_PUBLICADDRESS: "127.0.0.1:5701"
      HZ_NETWORK_RESTAPI_ENABLED: "true"
    image: "hazelcast/hazelcast:4.0.1"
    ports: 
      - "5701:5701"
  management-center: 
    container_name: management-center
    depends_on: 
      - hazelcast
    environment: 
      MC_ADMIN_PASSWORD: myPassword11
      MC_ADMIN_USER: admin
      MC_INIT_CMD: "./mc-conf.sh cluster add -H=/data -ma hazelcast:5701 -cn dev"
    image: "hazelcast/management-center:4.0.3"
    ports: 
      - "10080:8080"
version: "3"

hazelcast.xml - Data Connection Ayarları

Giriş

XML ile eklenen Data Connection bir daha silinemez

Örnek - JDBC

Şöyle yaparız

<hazelcast>
  <data-connection name="my-mysql-database">
    <type>JDBC</type>
    <properties>
      <property name="jdbcUrl">jdbc:mysql://mysql.example.org:3306</property> 
      <property name="user">my_user</property> 
      <property name="password">my_password</property>
    </properties>
    <shared>true</shared>
  </data-connection>
</hazelcast>

Örnek - Mongo

Şöyle yaparız

hazelcast:
  data-connection:
    my-mongodb:
      type: Mongo
      properties:
        connectionString: mongodb+srv://myusername:mypass@foo.mongodb.net/?retryWrites=true&w=majority
        database: hazelcast
        collection: supplements
      shared: true
  map:
    supplements:
      map-store:
        enabled: true
        class-name: com.hazelcast.mapstore.GenericMapStore
        properties:
          data-connection-ref: my-mongodb

Açıklaması şöyle

mongodb+srv: - The scheme that indicates this is a connection string for MongoDB Atlas.

myusername - The username for the user that will be connecting to the database.

mypass - The password for the user that will be connecting to the database.

foo.mongodb.net - The hostname of the MongoDB Atlas cluster.

retryWrites=true - Indicates that the driver should retry writes if they fail.

w=majority - Indicates that the driver should wait for a majority of nodes in the cluster to acknowledge a write before returning.

HZHibernate

Giriş

Bu proje idame ettiriliyor mu bilmiyorum. En son Hibernate 5.X sürümünde kalmış. Hibernate region factory olarak şu sınıflar kullanılabilir

1. HazelcastCacheRegionFactory

Şu satırı dahil ederiz

import com.hazelcast.hibernate.HazelcastCacheRegionFactory;

Second Level Cache verisini IMap üzerinde saklar. Bunu sağlayan sınıf IMapRegionCache

Kalıtım şöyle

org.hibernate.cache.spi.Region

  com.hazelcast.hibernate.RegionCache

    com.hazelcast.hibernate.distributed.IMapRegionCache

application.properties dosyasında şöyle yaparız

# HIBERNATE 2 LEVEL CACHE
# enable selective caching mode - only entities with @Cacheable annotation will use L2 cache
spring.jpa.properties.javax.persistence.sharedCache.mode=ENABLE_SELECTIVE
spring.jpa.properties.hibernate.cache.use_second_level_cache: true
spring.jpa.properties.hibernate.cache.region.factory_class: com.hazelcast.hibernate.HazelcastCacheRegionFactory

1 Key olarak org.hibernate.cache.internal.BasicCacheKeyImplementation kullanılıyor

2. Value olarak ta sanırım eğer kendi tarafımda value nesnesi yoksa "com.hazelcast.nio.serialization.HazelcastSerializationException : No DataSerializerFactory registered for namespace: ..." hatası alırım

2. HazelcastLocalCacheRegionFactory

Second  Level Cache verisini yerel bilgisayarda saklar. Bir şekilde arkada ITopic kullanıyor

Entity

Örnek

Şöyle yaparız. O zaman country isimli bir IMap oluşturur. Veri HeapData tipindendir

import org.hibernate.annotations.Cache;
import org.hibernate.annotations.CacheConcurrencyStrategy;
@Cache(usage = CacheConcurrencyStrategy.READ_WRITE, region="country")
public class Country {
   ...
}

Eğer region ismi vermezsek sınıfın tam yolunu kullanır. Yani IMap ismi IMap com.colak.hzspringtutorial.hibernatecache.Country gibi bir şey olur

3. JCache

Açıklaması şöyle

Hazelcast provides three CachingProvider implementations:

- A member-side implementation: the CacheManagers created by this provider are backed by member-side HazelcastInstances.

- A client-side implementation: the CacheManagers created by this provider are backed by client-side HazelcastInstances.

- A delegating caching provider that can be configured to delegate to the member- or client-side implementation.

Kalıtım şöyle

javax.cache.spi.CachingProvider
  // Client veya Member'a delegating Caching provider
  com.hazelcast.cache.HazelcastCachingProvider
  // Client'a delegating Caching provider
  com.hazelcast.client.cache.HazelcastClientCachingProvider
   // Member'a delegating Caching provider
  com.hazelcast.client.cache.HazelcastMemberCachingProvider

JCache CachingProvider gerçekleştirim ise şöyle

javax.cache.spi.CachingProvider
  com.hazelcast.cache.impl.AbstractHazelcastCachingProvider
    // com.hazelcast.client.cache.impl.HazelcastClientCacheManager yaratır
    com.hazelcast.client.cache.impl.HazelcastClientCachingProvider
    // com.hazelcast.client.cache.impl.HazelcastServerCacheManager yaratır
    com.hazelcast.cache.impl.HazelcastServerCachingProvider

Bu sınıflar JCache CacheManager nesneleri dönerler. Kalıtım şöyle

javax.cache.CacheManager
  com.hazelcast.cache.HazelcastCacheManager
    com.hazelcast.cache.impl.AbstractHazelcastCacheManager
      com.hazelcast.client.cache.impl.HazelcastClientCacheManager
      com.hazelcast.client.cache.impl.HazelcastServerCacheManager

Tüm CacheManager nesneleri altta aslında ICache arayüzünü kullanıyorlar

1. HazelcastClientCachingProvider - JCache

Örnek

Şöyle yaparız. Burada hazelcast.jcache.provider.type alanı member veya client değerlerini alabilir. member değeri atayarak yeni bir embedded member yaratması veya mevcut olanı kullanması sağlanıyor.

hazelcast.jcache.provider.type=member
hibernate.cache.region.factory_class=jcache
hibernate.cache.use_second_level_cache=true

2. HazelcastClientCachingProvider - JCache

Hazelcast cluster'a bağlanır

Örnek

Şöyle yaparız

jpa:
  default:
    properties:
      hibernate:
        cache:
          use_query_cache: false
          use_second_level_cache: true
          region:
            factory_class: com.hazelcast.client.cache.HazelcastClientCachingProvider

3. HazelcastServerCachingProvider - JCache

Embedded cache başlatır

Örnek

Şöyle yaparız

hibernate:
    cache:
      region:
        factory_class: jcache
    javax:
      cache:
        provider: com.hazelcast.cache.impl.HazelcastServerCachingProvider
        uri: <HazelcastConfig.xml>

hz-start komutu

Prometheus

Örnek

Şöyle yaparız

PROMETHEUS_PORT=8080 bin/hz-start

Micrometer HazelcastCacheMetrics Sınıfı

Örnek

Şöyle yaparız

import com.hazelcast.core.HazelcastInstance;
import io.micrometer.core.instrument.ImmutableTag;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.binder.cache.HazelcastCacheMetrics;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;


@Component
@RequiredArgsConstructor
public class HazelcastMetricBinder {
  public static String My_CACHE_NAME = "MyCacheName";
  private final MeterRegistry meterRegistry;
  private final HazelcastInstance hazelcastInstance;

  @PostConstruct
  void monitorHazelcastCacheMetrics() {
    List<Tag> tags = List.of(new ImmutableTag("cacheName", My_CACHE_NAME));
    HazelcastCacheMetrics.monitor(meterRegistry, 
                                  hazelcastInstance.getMap(My_CACHE_NAME),
                                  tags);
  }
}

hazelcast.xml - Metrics Ayarları

Giriş

Açıklaması şöyle

hazelcast.jmx enables MBeans that allow state changes, while hazelcast.metrics.jmx.enabled enables only read-only MBeans.

Örnek

Şöyle yaparız

  metrics:
    enabled: true
    jmx:
      enabled: true

Hazelcast Jet WriteKafkaP Sınıfı

Giriş
Şu satırı dahil ederiz 

import com.hazelcast.jet.kafka.impl.WriteKafkaP;

exactlyOnce Alanı

Kod şöyle. Producer sadece bir kere yazmayı garanti eder

public final class WriteKafkaP<T, K, V> implements Processor {

  ...
  private final boolean exactlyOnce;
}

Örnek

Şöyle yaparız. testJob çalışınca WriteKafkaP devreye girer

CREATE OR REPLACE MAPPING testMap (
  __key INT,
  ticker VARCHAR,
  price DECIMAL,
  amount BIGINT)
TYPE IMap
OPTIONS (
  'keyFormat'='int',
  'valueFormat'='json-flat'
);

INSERT INTO testMap VALUES
  (1, 'ABCD', 5.5, 10),
  (2, 'EFGH', 14, 20);

CREATE JOB testJob
OPTIONS (
  'processingGuarantee' = 'exactlyOnce'
) AS
SINK INTO testTopic
SELECT __key, ticker, price, amount FROM testMap

 private final boolean exactlyOnce;

HazelcastAPI CacheRecordStore Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.cache.impl.CacheRecordStore;

Kalıtım şöyle

ICacheRecordStore

  AbstractCacheRecordStore

    CacheRecordStore

Hazelcast Jet JsonUtil Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.json.JsonUtil;

beanFrom metodu

Örnek

Şöyle yaparız

Person person = JsonUtil.beanFrom("...", Person.class)

Hazelcast Jet DbField Sınıfı - Veri Tabanındaki Sütunu Temsil Eder

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.sql.impl.connector.jdbc.DbField;

DbField ve  MappingField İlişkisi

Örnek kod şöyle. MappingField "CREATE MAPPING ..." cümlesi ile belirtilen sütunu temsil eder. DbField ise veri tabanındaki sütunu temsil eder. DbField nesnesi QueryDataType nesnesine çevrilebilir.

private void validateType(MappingField field, DbField dbField) {
  QueryDataType type = resolveType(dbField.columnTypeName);
  if (!field.type().equals(type) && 
      !type.getConverter().canConvertTo(field.type().getTypeFamily())) {
    throw new IllegalStateException("Type " + 
      field.type().getTypeFamily() + 
      " of field " + field.name() +
      " does not match db type " + type.getTypeFamily());
  }
}