import com.hazelcast.jet.sql.impl.schema.TablesStorage;
boolean putIfAbsent(String name, Mapping mapping) {
return storage().putIfAbsent(name, mapping) == null;
}
IMap<String, Object> storage() {
return nodeEngine.getHazelcastInstance()
.getMap(JetServiceBackend.SQL_CATALOG_MAP_NAME);
}version: "2"services:postgres:image: postgres:12.4environment:- POSTGRES_USER=postgres- POSTGRES_PASSWORD=postgres- POSTGRES_DB=passwordports:- 5432:5432hazelcast:image: hazelcast/hazelcast:3.11ports:- 5701:5701environment:- JAVA_OPTS=-Dhazelcast.local.publicAddress=localhost:5701
version: '3.8'
name: HZ
services:
hazelcast1:
hostname: hazelcast1
image: hazelcast/hazelcast:5.3.2
deploy:
mode: replicated
replicas: 1
hazelcast2:
hostname: hazelcast2
image: hazelcast/hazelcast:5.3.2
deploy:
mode: replicated
replicas: 1
hazelcast3:
hostname: hazelcast3
image: hazelcast/hazelcast:5.3.2
deploy:
mode: replicated
replicas: 1
version: '3.8'
name : HZ
services:
hazelcast1:
hostname: hazelcast1
image: hazelcast/hazelcast:5.1.1-slim
environment:
HZ_NETWORK_JOIN_TCPIP_ENABLED: 'true'
HZ_NETWORK_JOIN_TCPIP_MEMBERS: hazelcast1,hazelcast2,hazelcast3
deploy:
mode: replicated
replicas: 1
hazelcast2:
hostname: hazelcast2
image: hazelcast/hazelcast:5.1.1-slim
environment:
HZ_NETWORK_JOIN_TCPIP_ENABLED: 'true'
HZ_NETWORK_JOIN_TCPIP_MEMBERS: hazelcast1,hazelcast2,hazelcast3
deploy:
mode: replicated
replicas: 1
hazelcast3:
hostname: hazelcast3
image: hazelcast/hazelcast:5.1.1-slim
environment:
HZ_NETWORK_JOIN_TCPIP_ENABLED: 'true'
HZ_NETWORK_JOIN_TCPIP_MEMBERS: hazelcast1,hazelcast2,hazelcast3
deploy:
mode: replicated
replicas: 1import com.hazelcast.jet.sql.impl.validate.HazelcastSqlValidator;
import com.hazelcast.jet.core.EventTimeMapper;
2. Some partition having no dataIt can happen that some partition does not have any events at all while others do, or the processor doesn't get any external partitions assigned to it. If we simply wait for the timestamps in all partitions to advance to some point, we won't be emitting any watermarks. This utility supports the idle timeout: if there's no new data from a partition after the timeout elapses, it will be marked as idle, allowing the processor's watermark to advance as if that partition didn't exist. If all partitions are idle or there are no partitions, the processor will emit a special idle message and the downstream will exclude this processor from watermark coalescing.
import com.hazelcast.config.QueueConfig;
String queueName = "event-queue"; QueueConfig queueConfig = config.getQueueConfig(queueName); queueConfig.setName(queueName) .setBackupCount(1) .setMaxSize(0) .setStatisticsEnabled(true);
Config hzConfig = hzInstance.getConfig(); QueueConfig iotQueueConfig = hzConfig.getQueueConfig(queueName); iotQueueConfig.setMaxSize(0); // unbounded
import com.hazelcast.jet.kafka.connect.impl.KafkaConnectSource;
Properties properties = new Properties();
properties.setProperty("name", "confluentinc-kafka-connect-jdbc");
properties.setProperty("connector.class", "io.confluent.connect.jdbc.JdbcSourceConnector");
properties.setProperty("mode", "incrementing");
properties.setProperty("tasks.max", "1");
String connectionUrl = ...
properties.setProperty("connection.url", connectionUrl);
properties.setProperty("connection.user", USERNAME);
properties.setProperty("connection.password", PASSWORD);
properties.setProperty("incrementing.column.name", "id");
properties.setProperty("table.whitelist", "dynamic_test_items1,dynamic_test_items2");
properties.setProperty("table.poll.interval.ms", "5000");
Pipeline pipeline = Pipeline.create();
StreamStage<String> streamStage = pipeline.readFrom(KafkaConnectSources.connect(properties))
.withoutTimestamps()
...import com.hazelcast.sql.impl.schema.Table;
Lite members are the Hazelcast cluster members that do not store data. These members are used mainly to execute tasks and register listeners and they do not have partitions.
A lite-member can become a full-member, but not the reverse.
import com.hazelcast.internal.cluster.ClusterService;
NodeEngineImpl nodeEngine; ClusterService clusterService = nodeEngine.getClusterService();
hz-cluster-cp-admin \-o remove-member \--member ${CPMEMBER_UUID} \--address 127.0.0.1 \--port 5701 \--groupname ${GROUPNAME} \--password ${PASSWORD}
... the default logging type of Hazelcast IMDG is JDK logging
<properties> <property name="hazelcast.logging.type">log4j2</property> /properties>
hazelcast-client:
...
properties:
hazelcast.logging.type: log4j2 @ClassRule public static final OverridePropertyRule enableLogging = set("hazelcast.logging.type", "log4j2");
<?xml version="1.0" encoding="UTF-8"?> <Configuration status="warn"> <Appenders> <!-- Console appender configuration --> <Console name="console" target="SYSTEM_OUT"> <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n"/> </Console> </Appenders> <Loggers> <!-- Root logger referring to console appender --> <Root level="info" additivity="false"> <AppenderRef ref="console"/> </Root> </Loggers> </Configuration>
System.setProperty("hazelcast.logging.type","log4j2");
System.setProperty("log4j2.configurationFile", "file:///.../mylog.properties");import com.hazelcast.jet.impl.connector.ReadJdbcP;
public ReadJdbcP(
SupplierEx<? extends Connection> newConnectionFn,
ToResultSetFunction resultSetFn,
FunctionEx<? super ResultSet, ? extends T> mapOutputFn
) {
...
}@Override
public boolean complete() {
if (traverser == null) {
resultSet = uncheckCall(() -> resultSetFn
.createResultSet(connection, parallelism, index));
traverser = ((Traverser<ResultSet>) () -> uncheckCall(() -> resultSet.next() ?
resultSet : null))
.map(mapOutputFn);
return emitFromTraverser(traverser);
}public static <T> ProcessorMetaSupplier supplier( SupplierEx<? extends DataSource> newDataSourceFn, ToResultSetFunction resultSetFn, FunctionEx<? super ResultSet, ? extends T> mapOutputFn ) public static <T> ProcessorMetaSupplier supplier( FunctionEx<ProcessorSupplier.Context, ? extends Connection> newConnectionFn, ToResultSetFunction resultSetFn, FunctionEx<? super ResultSet, ? extends T> mapOutputFn ) public static <T> ProcessorMetaSupplier supplier( String connectionURL, String query, FunctionEx<? super ResultSet, ? extends T> mapOutputFn ) public static <T> ProcessorMetaSupplier supplier( DataConnectionRef dataConnectionRef, ToResultSetFunction resultSetFn, FunctionEx<? super ResultSet, ? extends T> mapOutputFn)
import com.hazelcast.client.console.HazelcastCommandLine;
@Command(description = "Submits a job to the cluster")
public void submit(...) throws Exception {
...
HazelcastBootstrap.executeJar(
() -> getHazelcastClient(false),
file.getAbsolutePath(),
snapshotName,
name,
mainClass,
params,
false);
}import com.hazelcast.client.config.ClientSecurityConfig;
ClientConfig clientConfig = new ClientConfig(); clientConfig.setClusterName("hello-world"); clientConfig.getSecurityConfig().setUsernamePasswordIdentityConfig("member1","s3crEt"); HazelcastClient.newHazelcastClient(clientConfig);
import com.hazelcast.logging.ILogger; import com.hazelcast.logging.Logger;
private final ILogger logger = Logger.getLogger(Foo.class);
@Override
protected ILogger createLogger(String name) {
final Logger l = java.util.logging.Logger.getLogger(name);
return new StandardLogger(l);
}import com.hazelcast.jet.impl.JobProxy;
private <T> CompletableFuture<T> invokeOp(Operation op) {
return container()
.getOperationService()
.createInvocationBuilder(JetServiceBackend.SERVICE_NAME, op, coordinatorId())
.invoke();
}import com.hazelcast.client.impl.ClientEngine;
The ClientMessage class is the main message format used for communication between the client and members. This class encapsulates the message data and provides methods for encoding and decoding the data for transmission over the network.The ClientConnection class is responsible for managing the TCP socket connection between the client and a member. It handles sending and receiving messages, as well as managing the connection lifecycle (e.g. connecting, disconnecting, etc.).The ClientEngineImpl class is responsible for managing the overall client infrastructure, including maintaining a registry of connected clients and managing the communication between clients and members.
import com.hazelcast.client.impl.ClientEngineImpl;
public void accept(ClientMessage clientMessage) {
Connection connection = clientMessage.getConnection();
MessageTask messageTask = messageTaskFactory.create(clientMessage, connection);
OperationServiceImpl operationService = nodeEngine.getOperationService();
if (isUrgent(messageTask)) {
operationService.execute((UrgentMessageTask) messageTask);
} else if (messageTask instanceof AbstractPartitionMessageTask) {
operationService.execute((AbstractPartitionMessageTask) messageTask);
} else if (isQuery(messageTask)) {
queryExecutor.execute(messageTask);
} else if (messageTask instanceof TransactionalMessageTask) {
blockingExecutor.execute(messageTask);
} else if (messageTask instanceof BlockingMessageTask) {
blockingExecutor.execute(messageTask);
} else {
executor.execute(messageTask);
}
}import com.hazelcast.jet.impl.client.protocol.task.AbstractJetMessageTask;
@Override
protected InvocationBuilder getInvocationBuilder(Operation operation) {
Address address;
if (getLightJobCoordinator() != null) {
MemberImpl member = nodeEngine.getClusterService().getMember(getLightJobCoordinator());
if (member == null) {
throw new TopologyChangedException("Light job coordinator left the cluster");
}
address = member.getAddress();
} else {
address = nodeEngine.getMasterAddress();
if (address == null) {
throw new RetryableHazelcastException("master not yet known");
}
}
return nodeEngine.getOperationService()
.createInvocationBuilder(JetServiceBackend.SERVICE_NAME,
operation, address);
}It doesn't require you to add something like a class definition. You add serializers like you add a Portable factory, and versioning and schema distribution are handled for you
Compact serialization is the brand-new serialization mechanism built to have a small size on the serialized form yet allows fast data access for different query patterns. It is easy to use, language-independent, and performant.
It's worth noting that both Portable Serialization and Compact Serialization serve different use cases. Portable Serialization provides more flexibility, versioning support, and language interoperability, while Compact Serialization excels in minimizing the serialized data size and achieving high performance.
Compact serialization supports zero-configuration. In that case, it uses the fully qualified class name as its type name.
import com.hazelcast.nio.serialization.compact.CompactWriter;
<T> void writeCompact(String fieldName, T value);
import com.hazelcast.nio.serialization.genericrecord.GenericRecordBuilder;
import static com.hazelcast.nio.serialization.genericrecord.GenericRecordBuilder.compact;
import static com.hazelcast.nio.serialization.genericrecord.GenericRecordBuilder.portable;
GenericRecord genericRecord = ...
GenericRecordBuilder compactBuilder = compact("myrecord");
compactBuilder.setGenericRecord("record1", genericRecord);import com.hazelcast.internal.journal.EventJournalReader;
int partitionId = 1; CompletionStage completionStage1 = eventJournalReader.subscribeToEventJournal(partitionId); CompletionStage completionStage12 = eventJournalReader.readFromEventJournal(offset, 1, MAX_FETCH_SIZE, //128 partitionId, predicate, projection);