HazelcastAPI GenericMapLoader Sınıfı - Hazelcast ile Hazır Geliyor

Giriş

Şu satırı dahil ederiz

import com.hazelcast.mapStore.GenericMapLoader;

Cached Thread çalıştırır. Çıktısı şöyle

... cached.thread-5 - ... Dropping mapping __map-store....

Test Altyapısı - SimpleTestInClusterSupport Sınıfı

SimpleTestInClusterSupport

Bir çok test bu sınıftan direkt veya dolaylı olarak kalıtır. Kalıtım şöyle

HazelcastTestSupport

  JetTestSupport

    SimpleTestInClusterSupport

      SqlTestSupport

        JdbcSqlTestSupport: TestContainers Kullanarak Veri Tabanı Yaratır

          FooTest

Bu sınıfın için şöyle

public abstract class SimpleTestInClusterSupport extends JetTestSupport {
  ...
  private static TestHazelcastFactory factory;
  private static Config config;
  private static HazelcastInstance[] instances;
  private static HazelcastInstance client;
  ...

  @Nonnull
  protected static HazelcastInstance instance() {
    return instances[0];
  }

  /**
   * Returns all instances (except for the client).
   */
  @Nonnull
  protected static HazelcastInstance[] instances() {
    return instances;
  }

  /**
   * Returns the client or null, if a client wasn't requested.
   */
  protected static HazelcastInstance client() {
    return client;
  }
} 

initializeWithClient metodu

Belirtilen sayıda member ve bir tane de client yaratır

instances metodu

Şöyle yaparız

LogListener myListener = log -> ...
for (HazelcastInstance instance : instances()) {
  instance.getLoggingService().addLogListener(Level.FINE, myListener);
}

HazelcastClientAPI ListenerConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.ListenerConfig;

ClientConfig nesnesine Bir EventListener takar.

Örnek

Şöyle yaparız. com.hazelcast.core.LifecycleListener takılıyor

clientConfig.addListenerConfig(new ListenerConfig((LifecycleListener) event -> {
  if (event.getState().equals(CLIENT_CONNECTED)) {
    ...
  }
}));

Örnek

Şöyle yaparız. Böylece com.hazelcast.client.impl.management.ClientConnectionProcessListener nesnesinin possibleAddressesCollected(List<Address> addresses) metodu tetiklenir

ClientConnectionProcessListener listener = ...
ClientConfig clientConfig = ...
clientConfig.addListenerConfig(new ListenerConfig(listener));

HazelcastAPI ClusterProperty Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.spi.properties.ClusterProperty;

Örnek

Şöyle yaparız. Bu cluster her bir RPC için 200 milisaniyede cevap bekliyor.

Config config = ...
config.setProperty(ClusterProperty.OPERATION_CALL_TIMEOUT_MILLIS.getName(), "200";

HazelcastClientAPI ClientProperty Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.client.properties.ClientProperty;

Örnek

Şöyle yaparız. Bu client her bir RPC isteği için 10 saniyede cevap bekliyor.

ClientConfig clientConfig = new ClientConfig();
clientConfig.setProperty(ClientProperty.INVOCATION_TIMEOUT_SECONDS.getName(), "10");

Hazelcast SQL - DELETE FROM

Söz dizimi şöyle

DELETE FROM <mapping name> [where condition];

Docker Extension

GitHub bağlantısı burada

https://github.com/hazelcast/hazelcast-docker-extension

Altta Management Center'ı çalıştırır ve onun çıktısını gösterir

HazelcastAPI TcpIpConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.TcpIpConfig;

setConnectionTimeoutSeconds

Örnek

Şöyle yaparız

private NetworkConfig getHazelcastNetworkConfig(){

  NetworkConfig networkConfig = new NetworkConfig().setPort(5900)  
    .setPortAutoIncrement(false);

  JoinConfig joinConfig = new JoinConfig();
  TcpIpConfig tcpIpConfig = new TcpIpConfig();
  tcpIpConfig.setConnectionTimeoutSeconds(30); 
  tcpIpConfig.setEnabled(true);
    
  List<String> memberList = new ArrayList<>();
  memberList.add("127.0.0.1:5900");
  tcpIpConfig.setMembers(memberList);

  joinConfig.setTcpIpConfig(tcpIpConfig);

  networkConfig.setJoin(joinConfig);

  return networkConfig;
}

setMembers metodu

Örnek

Şöyle yaparız

@Bean("hazelcastInstanceTest")
public HazelcastInstance hazelcastInstance() {
  Config config = new Config();
  config.getNetworkConfig().addOutboundPort(0);

  JoinConfig joinConfig = config.getNetworkConfig().getJoin();

  joinConfig.getMulticastConfig()
    .setEnabled(false);

  List<String> members = Arrays.asList("MEMBER1:5701","MEMBER2:5701");
  joinConfig.getTcpIpConfig()
    .setMembers(members)
    .setEnabled(true);

  return Hazelcast.newHazelcastInstance(config);
}

Hazelcast SQL - DROP MAPPING

Örnek

Şöyle yaparız.

DROP MAPPING "Patient"

Örnek

Şöyle yaparız.

DROP MAPPING IF EXISTS "Patient"

HazelcastClientAPI ClientConnectionStrategyConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.client.config.ClientConnectionStrategyConfig;

getConnectionRetryConfig metodu

ConnectionRetryConfig döner

setAsyncStart metodu

Örnek

Şöyle yaparız.

ClientConfig clientConfig = new ClientConfig();
...
ClientConnectionStrategyConfig connectionStrategyConfig = clientConfig
  .getConnectionStrategyConfig();
connectionStrategyConfig
  .setAsyncStart(true)
  .setReconnectMode(ClientConnectionStrategyConfig.ReconnectMode.ASYNC);
  ...
HazelcastInstance clientInstance = HazelcastClient.newHazelcastClient(clientConfig);

setReconnectMode metodu

İmzası şöyle

public ClientConnectionStrategyConfig setReconnectMode(ReconnectMode reconnectMode)

ClientConnectionStrategyConfig.ReconnectMode.ON , ASYNC , OFF değerleri verilebilir. Açıklaması şöyle

... the client will try to reconnect both in the ON and ASYNC modes. The difference comes from the fate of the pending invocations during that time. With the ON mode, the requests are blocked i.e. you don't get a result back for your API calls until the client reconnects. With the ASYNC mode, requests are failed immediately with the HazelcastClientOfflineException.

HazelcastClientAPI ConnectionRetryConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.client.config.ConnectionRetryConfig;

setClusterConnectTimeoutMillis metodu

Cluster'a bağlanma zaman aşımı süresini belirtir.

Örnek

Şöyle yaparız.

ClientConfig clientConfig = new ClientConfig();
...
ClientConnectionStrategyConfig connectionStrategyConfig = clientConfig
  .getConnectionStrategyConfig();
connectionStrategyConfig
  .setAsyncStart(true)
  .setReconnectMode(ClientConnectionStrategyConfig.ReconnectMode.ASYNC);

ConnectionRetryConfig connectionRetryConfig = connectionStrategyConfig
  .getConnectionRetryConfig();
connectionRetryConfig
  .setInitialBackoffMillis(1_000)
  .setMaxBackoffMillis(60_000)
  .setMultiplier(2)
  .setClusterConnectTimeoutMillis(50_000)
  .setJitter(0.2);

HazelcastInstance clientInstance = HazelcastClient.newHazelcastClient(clientConfig);

Hazelcast Jet PulsarSources Sınıfı

Maven

Şu satırı dahil ederiz

<dependency>
  <groupId>com.hazelcast</groupId>
  <artifactId>hazelcast</artifactId>
  <version>5.1.4</version>
</dependency>

<dependency>
  <groupId>com.hazelcast.jet.contrib</groupId>
  <artifactId>pulsar</artifactId>
  <version>0.1</version>
</dependency>

<dependency>
  <groupId>org.apache.pulsar</groupId>
  <artifactId>pulsar-client</artifactId>
  <version>2.10.1</version>
</dependency>

Örnek

Şöyle yaparız

StreamSource<Event>source = PulsarSources.pulsarReaderBuilder(
  topicName,
  () -> PulsarClient.builder().serviceUrl(“pulsar://localhost:6650”).build(),
  () -> Schema.JSON(Event.class),
  Message::getValue)
.build();

Pipeline p = Pipeline.create();
p.readFrom(source)
 .withNativeTimestamps(0)
 .groupingKey(Event::getUser)
 .window(sliding(SECONDS.toMillis(60), SECONDS.toMillis(30)))
 .aggregate(counting())
 .writeTo(Sinks.logger(wr -> String.format(
    “At %s Pulsar got %,d messages in the previous minute from %s.”,
      TIME_FORMATTER.format(LocalDateTime.ofInstant(
        Instant.ofEpochMilli(wr.end()), ZoneId.systemDefault())),
      wr.result(), wr.key())));

JobConfig cfg = new JobConfig()
  .setProcessingGuarantee(ProcessingGuarantee.EXACTLY_ONCE)
  .setSnapshotIntervalMillis(SECONDS.toMillis(1))
  .setName(“pulsar-airquality-counter”);

HazelcastInstance hz = Hazelcast.bootstrappedInstance();
hz.getJet().newJob(p, cfg);

hz-mc komutu - Management Center İçindir

-V seçeneği

Şöyle yaparız

hz-mc -V

hz komutu - Start the Member

Kurulum

Şöyle yaparız

brew tap hazelcast/hz

brew install hazelcast@5.2.1

Dosya burada

PROMETHEUS

Kod şöyle

if [ "${PROMETHEUS_PORT}" ]; then
  echo "Prometheus enabled on port ${PROMETHEUS_PORT}"
  PROMETHEUS="-javaagent:${HAZELCAST_HOME}/lib/jmx_prometheus_javaagent-${prometheus.version}.jar=${PROMETHEUS_PORT}:${HAZELCAST_HOME}/config/jmx_agent_config.yaml"
fi

Bu aslında jmx_exporter isimli bir java agent çalıştırıyor. Eğer java agent'i tek başına çalıştırmak istersek şöyle yaparız 

java -javaagent:./jmx_prometheus_javaagent-0.19.0.jar=12345:config.yaml -jar yourJar.jar

Ama Hazelcast için distribution/pom.xml dosyasında PROMETHEUS JMS Exporter dahil ediliyor. Bu bağımlılık lib dizinine taşınıyor.

<!-- Prometheus to expose JMX metrics as HTTP endpoint -->
<dependency>
  <groupId>io.prometheus.jmx</groupId>
  <artifactId>jmx_prometheus_javaagent</artifactId>
  <version>${prometheus.version}</version>
</dependency>

start seçeneği

Örnek

Şöyle yaparız

hz start

- V seçeneği

Şöyle yaparız

hz -V

HazelcastAPI JoinConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.JoinConfig;

Örnek - Network (Ağ Bağlantısı) Olmasın

Şöyle yaparız. TCP ve Multicast etkisiz hale getiriliyor.

Config config = new Config();
config.setProperty("hazelcast.shutdownhook.enabled", "false");
NetworkConfig network = config.getNetworkConfig();
network.getJoin().getTcpIpConfig().setEnabled(false);
network.getJoin().getMulticastConfig().setEnabled(false);
HazelcastInstance instance = Hazelcast.newHazelcastInstance(config);

getKubernetesConfig metodu

KubernetesConfig nesnesi döner

getTcpIpConfig metodu

TcpIpConfig döner

setMulticastConfig metodu

Örnek

Şöyle yaparız

private NetworkConfig getHazelcastNetworkConfig(){

  NetworkConfig networkConfig = new NetworkConfig().setPort(5900)  
    .setPortAutoIncrement(false);

  JoinConfig joinConfig = new JoinConfig();
    

  MulticastConfig multicastConfig = new MulticastConfig();
  multicastConfig.setMulticastTimeoutSeconds(30);
  multicastConfig.setMulticastTimeToLive(255);
  multicastConfig.setEnabled(false);

  joinConfig.setMulticastConfig(multicastConfig);

  networkConfig.setJoin(joinConfig);

  return networkConfig;
}

setTcpIpConfig metodu

TcpIpConfig ekler

HazelcastAPI KubernetesConfig Sınıfı - Kubernetes Join İçindir

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.KubernetesConfig;

setProperty metodu

namespace ve service-name değerlerini atamak gerekir

Örnek

Şöyle yaparız

Config config = getCommonConfig();
JoinConfig joinConfig = config.getNetworkConfig().getJoin();
joinConfig.getMulticastConfig().setEnabled(false);
joinConfig.getKubernetesConfig().setEnabled(true)
        .setProperty("namespace", namespace)
        .setProperty("service-name", service);

Örnek

Şöyle yaparız

Config config = new Config();
JoinConfig joinConfig = config.getNetworkConfig().getJoin();
joinConfig.getTcpIpConfig().setEnabled(false);
joinConfig.getMulticastConfig().setEnabled(false);

config
  .setClusterName("hz-example")
  .getNetworkConfig()
    .setPort(5701)
    .addOutboundPortDefinition("20000-30000")
  .getJoin()
    .getKubernetesConfig()
      .setEnabled(true)
      .setProperty("namespace", "example_ns")
      .setProperty("service-name", "example_service")
      .setProperty("service-port", "5701");

Cached Thread

Giriş

İsmi cached.thread-N şeklindedir. Şöyle yaratılır

public final class ExecutionServiceImpl implements ExecutionService {

  private final ExecutorService cachedExecutorService;
  ...
}

Stack trace şöyle

run:-1, Foo$$Lambda$1822/0x00000008012739a0 
call:539, Executors$RunnableAdapter (java.util.concurrent)
run:64, CompletableFutureTask (com.hazelcast.internal.util.executor)
run:217, CachedExecutorServiceDelegate$Worker (com.hazelcast.internal.util.executor)
runWorker:1136, ThreadPoolExecutor (java.util.concurrent)
run:635, ThreadPoolExecutor$Worker (java.util.concurrent)
run:833, Thread (java.lang)
executeRun:76, HazelcastManagedThread (com.hazelcast.internal.util.executor)
run:111, HazelcastManagedThread (com.hazelcast.internal.util.executor)

HazelcastAPI SerializerConfig Sınıfı - Belirli Bir Sınıf İçin Serialization Ayarlarını Belirtir

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.SerializerConfig;

setTypeClass metodu

Örnek

Şöyle yaparız

config.getSerializationConfig()
  .addSerializerConfig(new SerializerConfig()
                             .setTypeClass(CartMapping.class)
                             .setImplementation(cartMappingSerializer()));

Operation Thread - Fixed Java Pool (FJP)

Giriş

Kısaca iki tane farklı ortam değişkeni var. Bunlar şöyle

hazelcast.operation.generic.thread.count // executor service
hazelcast.operation.thread.count // map get put

1. Non-Partition-aware Operations

Açıklaması şöyle. Core sayısının yarısı veya en az 2 tane yaratılır. 

To execute operations that are not partition-aware, e.g., IExecutorService.executeOnMember(command, member), generic operation threads are used. When the Hazelcast instance is started, an array of operation threads is created. The size of this array has a default value of the number of cores divided by two with a minimum value of 2. It can be changed using the hazelcast.operation.generic.thread.count property.

2. Partition-aware operations

Açıklaması şöyle

These threads execute Hazelcast operations, such as map put or get operations

Açıklaması şöyle. Core sayısı kadar veya en az 2 tane Partition Thread yaratılır. 

To execute partition-aware operations, an array of operation threads is created. The default value of this array’s size is the number of cores and it has a minimum value of 2. This value can be changed using the hazelcast.operation.thread.count property.

Each operation thread has its own work queue and it consumes messages from this work queue. If a partition-aware operation needs to be scheduled, the right thread is found using the formula below.

threadIndex = partitionId % partition thread-count

After the threadIndex is determined, the operation is put in the work queue of that operation thread. This means the followings:

• A single operation thread executes operations for multiple partitions; if there are 271 partitions and 10 partition threads, then roughly every operation thread executes operations for 27 partitions.
• Each partition belongs to only 1 operation thread. All operations for a partition are always handled by exactly the same operation thread.
• Concurrency control is not needed to deal with partition-aware operations because once a partition-aware operation is put in the work queue of a partition-aware operation thread, only 1 thread is able to touch that partition.

Örnek - XML

Şöyle yaparız

<hazelcast>
  <properties>
    <property name="hazelcast.operation.thread.count">10</property>
  </properties>
  ...
</hazelcast>

2.1 Nasıl Çalışır

Açıklaması şöyle. 

Each operation thread has its own work queue and it consumes messages from this work queue. If a partition-aware operation needs to be scheduled, the right thread is found using the formula below.

threadIndex = partitionId % partition thread-count

After the threadIndex is determined, the operation is put in the work queue of that operation thread. This means the followings:

• A single operation thread executes operations for multiple partitions; if there are 271 partitions and 10 partition threads, then roughly every operation thread executes operations for 27 partitions.
• Each partition belongs to only 1 operation thread. All operations for a partition are always handled by exactly the same operation thread.
• Concurrency control is not needed to deal with partition-aware operations because once a partition-aware operation is put in the work queue of a partition-aware operation thread, only 1 thread is able to touch that partition.

2.2 Partition Thread İsmi

hz.XXX.partition-operation.thread-N şeklinde. Eğer örneğin 16 işlemci varsa son kısım, thread-0 ve thread-15 arasında bir şey olur

hz.XXX.partition-operation.thread-0

Örnek

Put işlemi için çağrı şöyle. Buraya kadar her şey partition-operation.thread üzerinde çalışıyor

interceptPut:641, MapServiceContextImpl (com.hazelcast.map.impl)
putInternal:946, DefaultRecordStore (com.hazelcast.map.impl.recordstore)
putIfAbsent:1218, DefaultRecordStore (com.hazelcast.map.impl.recordstore)
runInternal:39, PutIfAbsentOperation (com.hazelcast.map.impl.operation)
run:153, MapOperation (com.hazelcast.map.impl.operation)
call:173, MapOperation (com.hazelcast.map.impl.operation)
call:295, OperationRunnerImpl (com.hazelcast.spi.impl.operationservice.impl)
run:270, OperationRunnerImpl (com.hazelcast.spi.impl.operationservice.impl)
run:219, OperationRunnerImpl (com.hazelcast.spi.impl.operationservice.impl)
process:180, OperationThread (com.hazelcast.spi.impl.operationexecutor.impl)
process:144, OperationThread (com.hazelcast.spi.impl.operationexecutor.impl)
loop:134, OperationThread (com.hazelcast.spi.impl.operationexecutor.impl)
executeRun:115, OperationThread (com.hazelcast.spi.impl.operationexecutor.impl)
run:111, HazelcastManagedThread (com.hazelcast.internal.util.executor)

hazelcast.xml - Map Index Ayarları

Index Type 

Index tipleri şöyle

SORTED (default): For ranged queries such as listing all employees that are between 40 and 60.

HASH: For unordered queries such as getting the IDs of employees that are named John.

BITMAP: For querying fields that contain few distinct values such as boolean fields. See Bitmap Indexes.

Açıklaması şöyle

The this keyword acts on the value of a map entry to access properties on it. For example, if a value contains an employee object with the name field, you can access that field, using this.name. Typically, you do not need to specify the this keyword because its presence is assumed if the special attribute __key is not specified. As a result, this.name and name are equivalent.

VALUE NESNESİNE INDEX

İndekste value nesnesinin alanının ismi attribute olarak belirtilir.

Örnek

Şöyle yaparız. Burada value nesnesinin field1 ve field2 isimli iki tane alanı var.

<map name="indexedMap">
  <in-memory-format>NATIVE</in-memory-format>

  <indexes>
    <index type="HASH">
      <attributes>
        <attribute>field1</attribute>
      </attributes>
    </index>
    <index type="HASH">
      <attributes>
        <attribute>field2</attribute>
      </attributes>
    </index>
  </indexes>
</map>


IMap<Integer, FooValue> map = client.getMap("indexedMap");
IntStream.range(0, 5_000_000).forEach(i -> {
  map.put(i, new FooValue("field1" + 1, "field2" + i));
  map.remove(i);
});

HASH

Örnek 

Şöyle yaparız. Burada value nesnesinin component ve instant isimli iki tane alanı var.

hazelcast:
  map:
    analytics:
      indexes:
        - type: HASH
          attributes:
            - "component"
        - type: SORTED
          attributes:
            - "instant"

SORTED

Index tipine SORTED kelimesini açıktan yazmya gerek yok

Örnek

Şöyle yaparız. Bu sefer index tipine SORTED açıktan yazılmıyor. Burada value nesnesinin name ve age isimli iki tane alanı var.

<hazelcast>
  ...
  <map name="employees">
    <indexes>
      <index type="HASH">
        <attributes>
          <attribute>name</attribute>
        </attributes>
      </index>
      <index>
        <attributes>
          <attribute>age</attribute>
        </attributes>
      </index>
    </indexes>
  </map>
  ...
</hazelcast>

Key Nesnesine Index

Örnek

Key nesnesinin bir alanına indeks koymak için şöyle yaparız. Burada key nesnesinin eventType isimli bir alanı var

<indexes>
  <index>
    <attributes>
      <attribute>__key.eventType</attribute>
    </attributes>
  </index>
</indexes>

Hazelcast SQL - Mongo İçin CREATE MAPPING Örnekleri

Örnek

Şöyle yaparız. Burada mapping name ile database + collection name farklı

CREATE MAPPING people EXTERNAL NAME "database"."collectionName" (
    id OBJECT EXTERNAL NAME _id
    firstName VARCHAR,
    lastName VARCHAR, 
    employed BOOLEAN 
)
TYPE Mongo
OPTIONS (
  'connectionString' = 'mongodb://127.0.0.1:27017'
);

Örnek

Şöyle yaparız

CREATE MAPPING mydetails (
  id OBJECT EXTERNAL NAME _id,
  first_name VARCHAR,
  last_name VARCHAR
)
TYPE Mongo
OPTIONS (
  'connectionString' = 'mongodb://127.0.0.1:27017',
);

Job Upload

Giriş

client bir jar yükleyip JetJob olarak çalıştırabilir. 

client member üzerindeki bir jar'ı direkt çalıştırabilir.

execute

Akış şöyle. 

# storeJobMetaData
UploadJobMetaDataOperation  -> JetServiceBackend
                                         |
                                         .executeJar -> HazelcastBootstrap 

upload 

Akış şöyle. Yani JetServiceBackend içinde bir tane JobUploadStore nesnesi var.

# storeJobMetaData
UploadJobMetaDataOperation  -> JetServiceBackend -> JobUploadStore.processJobMetaData

# storeJobMultiPart
UploadJobMultiPartOperation -> JetServiceBackend -> JobUploadStore.processJobMultipart
                                         |
                                         .executeJar -> HazelcastBootstrap 

Hazelcast SQL INNER JOIN

Örnek

Şöyle yaparız

SELECT b.* FROM Booking b 
 INNER JOIN Users u ON b.salesManagerId = u.icNumber 
 AND u.dealerCode = 'B129' 
 AND u.status = 1;

Hazelcast Jet DebeziumCdcSources Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.cdc.DebeziumCdcSources;

Örnek

Şöyle yaparız

StreamSource<ChangeRecord> source = DebeziumCdcSources
  .debezium("postgres", PostgresConnector.class)
  .setProperty("database.server.name",...)
  .setProperty("database.hostname", ...)
  .setProperty("database.port", ...)
  .setProperty("database.user", ...)
  .setProperty("database.password", ...)
  .setProperty("database.dbname", ...)
  .setProperty("table.whitelist", ...)
  .build();


Pipeline pipeline = Pipeline.create();
pipeline.readFrom(source)
  .withNativeTimestamps(1)
  .writeTo(Sinks.list("no_pk"));