You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/11/21 09:48:31 UTC
[GitHub] [flink-connector-cassandra] echauchot opened a new pull request, #1: [FLINK-29982] move the existing Cassandra connector to external repository
echauchot opened a new pull request, #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1
The migration process states that we must have 1 release of Flink with the connector being released both separately and as part of Flink. So this PR is the copy of the existing Cassandra connector (not implementing the new Source API) from the Flink repo to new dedicated Cassandra repo:
R: @zentol
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] zentol commented on a diff in pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#discussion_r1027825022
##########
src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java:
##########
@@ -0,0 +1,912 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.cassandra;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo;
+import org.apache.flink.batch.connectors.cassandra.CassandraInputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraPojoInputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraPojoOutputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraRowOutputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraTupleOutputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
+import org.apache.flink.streaming.runtime.operators.WriteAheadSinkTestBase;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.testutils.junit.RetryOnException;
+import org.apache.flink.testutils.junit.extensions.retry.RetryExtension;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.DockerImageVersions;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SimpleStatement;
+import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.Statement;
+import com.datastax.driver.core.exceptions.NoHostAvailableException;
+import com.datastax.driver.mapping.Mapper;
+import com.datastax.driver.mapping.annotations.Table;
+import net.bytebuddy.ByteBuddy;
+import org.assertj.core.api.recursive.comparison.RecursiveComparisonConfiguration;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.CassandraContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.io.IOException;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Constructor;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+
+import scala.collection.JavaConverters;
+import scala.collection.Seq;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for all cassandra sinks. */
+@SuppressWarnings("serial")
+// NoHostAvailableException is raised by Cassandra client under load while connecting to the cluster
+@RetryOnException(times = 3, exception = NoHostAvailableException.class)
+@Testcontainers
+@ExtendWith(RetryExtension.class)
+class CassandraConnectorITCase
+ extends WriteAheadSinkTestBase<
+ Tuple3<String, Integer, Integer>,
+ CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>>> {
+
+ private static final int MAX_CONNECTION_RETRY = 3;
+ private static final long CONNECTION_RETRY_DELAY = 500L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(CassandraConnectorITCase.class);
+ private static final Slf4jLogConsumer LOG_CONSUMER = new Slf4jLogConsumer(LOG);
+
+ @TempDir static Path tmpDir;
+
+ private static final int READ_TIMEOUT_MILLIS = 36000;
+
+ @Container static final CassandraContainer CASSANDRA_CONTAINER = createCassandraContainer();
+
+ private static final int PORT = 9042;
+
+ private static Cluster cluster;
+ private static Session session;
+
+ private final ClusterBuilder builderForReading =
+ createBuilderWithConsistencyLevel(ConsistencyLevel.ONE);
+ // Lower consistency level ANY is only available for writing.
+ private final ClusterBuilder builderForWriting =
+ createBuilderWithConsistencyLevel(ConsistencyLevel.ANY);
+
+ private ClusterBuilder createBuilderWithConsistencyLevel(ConsistencyLevel consistencyLevel) {
+ return new ClusterBuilder() {
+ @Override
+ protected Cluster buildCluster(Cluster.Builder builder) {
+ return builder.addContactPointsWithPorts(
+ new InetSocketAddress(
+ CASSANDRA_CONTAINER.getHost(),
+ CASSANDRA_CONTAINER.getMappedPort(PORT)))
+ .withQueryOptions(
+ new QueryOptions()
+ .setConsistencyLevel(consistencyLevel)
+ .setSerialConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL))
+ .withSocketOptions(
+ new SocketOptions()
+ // default timeout x 3
+ .setConnectTimeoutMillis(15000)
+ // default timeout x3 and higher than
+ // request_timeout_in_ms at the cluster level
+ .setReadTimeoutMillis(READ_TIMEOUT_MILLIS))
+ .withoutJMXReporting()
+ .withoutMetrics()
+ .build();
+ }
+ };
+ }
+
+ private static final String TABLE_NAME_PREFIX = "flink_";
+ private static final String TABLE_NAME_VARIABLE = "$TABLE";
+ private static final String KEYSPACE = "flink";
+ private static final String TUPLE_ID_FIELD = "id";
+ private static final String TUPLE_COUNTER_FIELD = "counter";
+ private static final String TUPLE_BATCHID_FIELD = "batch_id";
+ private static final String CREATE_KEYSPACE_QUERY =
+ "CREATE KEYSPACE "
+ + KEYSPACE
+ + " WITH replication= {'class':'SimpleStrategy', 'replication_factor':1};";
+ private static final String CREATE_TABLE_QUERY =
+ "CREATE TABLE "
+ + KEYSPACE
+ + "."
+ + TABLE_NAME_VARIABLE
+ + " ("
+ + TUPLE_ID_FIELD
+ + " text PRIMARY KEY, "
+ + TUPLE_COUNTER_FIELD
+ + " int, "
+ + TUPLE_BATCHID_FIELD
+ + " int);";
+ private static final String INSERT_DATA_QUERY =
+ "INSERT INTO "
+ + KEYSPACE
+ + "."
+ + TABLE_NAME_VARIABLE
+ + " ("
+ + TUPLE_ID_FIELD
+ + ", "
+ + TUPLE_COUNTER_FIELD
+ + ", "
+ + TUPLE_BATCHID_FIELD
+ + ") VALUES (?, ?, ?)";
+ private static final String SELECT_DATA_QUERY =
+ "SELECT * FROM " + KEYSPACE + "." + TABLE_NAME_VARIABLE + ';';
+
+ private static final Random random = new Random();
+ private int tableID;
+
+ private static final ArrayList<Tuple3<String, Integer, Integer>> collection =
+ new ArrayList<>(20);
+ private static final ArrayList<Row> rowCollection = new ArrayList<>(20);
+
+ private static final TypeInformation[] FIELD_TYPES = {
+ BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO
+ };
+
+ static {
+ for (int i = 0; i < 20; i++) {
+ collection.add(new Tuple3<>(UUID.randomUUID().toString(), i, 0));
+ rowCollection.add(Row.of(UUID.randomUUID().toString(), i, 0));
+ }
+ }
+
+ private static Class<? extends Pojo> annotatePojoWithTable(String keyspace, String tableName) {
+ return new ByteBuddy()
+ .redefine(Pojo.class)
+ .name("org.apache.flink.streaming.connectors.cassandra.Pojo" + tableName)
+ .annotateType(createTableAnnotation(keyspace, tableName))
+ .make()
+ .load(Pojo.class.getClassLoader())
+ .getLoaded();
+ }
+
+ @NotNull
+ private static Table createTableAnnotation(String keyspace, String tableName) {
+ return new Table() {
+
+ @Override
+ public String keyspace() {
+ return keyspace;
+ }
+
+ @Override
+ public String name() {
+ return tableName;
+ }
+
+ @Override
+ public boolean caseSensitiveKeyspace() {
+ return false;
+ }
+
+ @Override
+ public boolean caseSensitiveTable() {
+ return false;
+ }
+
+ @Override
+ public String writeConsistency() {
+ return "";
+ }
+
+ @Override
+ public String readConsistency() {
+ return "";
+ }
+
+ @Override
+ public Class<? extends Annotation> annotationType() {
+ return Table.class;
+ }
+ };
+ }
+
+ // ------------------------------------------------------------------------
+ // Utility methods
+ // ------------------------------------------------------------------------
+
+ public static CassandraContainer createCassandraContainer() {
+ CassandraContainer cassandra = new CassandraContainer(DockerImageVersions.CASSANDRA_4_0);
Review Comment:
Don't rely on the Flink DockerImageVersions; this entry will be removed eventually.
##########
pom.xml:
##########
@@ -0,0 +1,432 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>io.github.zentol.flink</groupId>
+ <artifactId>flink-connector-parent</artifactId>
+ <version>1.0</version>
+ </parent>
+
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-cassandra_${scala.binary.version}</artifactId>
+ <version>4.0-SNAPSHOT</version>
Review Comment:
```suggestion
<version>3.0-SNAPSHOT</version>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] zentol commented on pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#issuecomment-1321849479
We'd like to preserve the git history; can you rebase your PR against the `cas_rewrite` branch in this repo?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot commented on a diff in pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot commented on code in PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#discussion_r1028076179
##########
src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java:
##########
@@ -0,0 +1,912 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.cassandra;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo;
+import org.apache.flink.batch.connectors.cassandra.CassandraInputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraPojoInputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraPojoOutputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraRowOutputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraTupleOutputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
+import org.apache.flink.streaming.runtime.operators.WriteAheadSinkTestBase;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.testutils.junit.RetryOnException;
+import org.apache.flink.testutils.junit.extensions.retry.RetryExtension;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.DockerImageVersions;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SimpleStatement;
+import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.Statement;
+import com.datastax.driver.core.exceptions.NoHostAvailableException;
+import com.datastax.driver.mapping.Mapper;
+import com.datastax.driver.mapping.annotations.Table;
+import net.bytebuddy.ByteBuddy;
+import org.assertj.core.api.recursive.comparison.RecursiveComparisonConfiguration;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.CassandraContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.io.IOException;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Constructor;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+
+import scala.collection.JavaConverters;
+import scala.collection.Seq;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for all cassandra sinks. */
+@SuppressWarnings("serial")
+// NoHostAvailableException is raised by Cassandra client under load while connecting to the cluster
+@RetryOnException(times = 3, exception = NoHostAvailableException.class)
+@Testcontainers
+@ExtendWith(RetryExtension.class)
+class CassandraConnectorITCase
+ extends WriteAheadSinkTestBase<
+ Tuple3<String, Integer, Integer>,
+ CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>>> {
+
+ private static final int MAX_CONNECTION_RETRY = 3;
+ private static final long CONNECTION_RETRY_DELAY = 500L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(CassandraConnectorITCase.class);
+ private static final Slf4jLogConsumer LOG_CONSUMER = new Slf4jLogConsumer(LOG);
+
+ @TempDir static Path tmpDir;
+
+ private static final int READ_TIMEOUT_MILLIS = 36000;
+
+ @Container static final CassandraContainer CASSANDRA_CONTAINER = createCassandraContainer();
+
+ private static final int PORT = 9042;
+
+ private static Cluster cluster;
+ private static Session session;
+
+ private final ClusterBuilder builderForReading =
+ createBuilderWithConsistencyLevel(ConsistencyLevel.ONE);
+ // Lower consistency level ANY is only available for writing.
+ private final ClusterBuilder builderForWriting =
+ createBuilderWithConsistencyLevel(ConsistencyLevel.ANY);
+
+ private ClusterBuilder createBuilderWithConsistencyLevel(ConsistencyLevel consistencyLevel) {
+ return new ClusterBuilder() {
+ @Override
+ protected Cluster buildCluster(Cluster.Builder builder) {
+ return builder.addContactPointsWithPorts(
+ new InetSocketAddress(
+ CASSANDRA_CONTAINER.getHost(),
+ CASSANDRA_CONTAINER.getMappedPort(PORT)))
+ .withQueryOptions(
+ new QueryOptions()
+ .setConsistencyLevel(consistencyLevel)
+ .setSerialConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL))
+ .withSocketOptions(
+ new SocketOptions()
+ // default timeout x 3
+ .setConnectTimeoutMillis(15000)
+ // default timeout x3 and higher than
+ // request_timeout_in_ms at the cluster level
+ .setReadTimeoutMillis(READ_TIMEOUT_MILLIS))
+ .withoutJMXReporting()
+ .withoutMetrics()
+ .build();
+ }
+ };
+ }
+
+ private static final String TABLE_NAME_PREFIX = "flink_";
+ private static final String TABLE_NAME_VARIABLE = "$TABLE";
+ private static final String KEYSPACE = "flink";
+ private static final String TUPLE_ID_FIELD = "id";
+ private static final String TUPLE_COUNTER_FIELD = "counter";
+ private static final String TUPLE_BATCHID_FIELD = "batch_id";
+ private static final String CREATE_KEYSPACE_QUERY =
+ "CREATE KEYSPACE "
+ + KEYSPACE
+ + " WITH replication= {'class':'SimpleStrategy', 'replication_factor':1};";
+ private static final String CREATE_TABLE_QUERY =
+ "CREATE TABLE "
+ + KEYSPACE
+ + "."
+ + TABLE_NAME_VARIABLE
+ + " ("
+ + TUPLE_ID_FIELD
+ + " text PRIMARY KEY, "
+ + TUPLE_COUNTER_FIELD
+ + " int, "
+ + TUPLE_BATCHID_FIELD
+ + " int);";
+ private static final String INSERT_DATA_QUERY =
+ "INSERT INTO "
+ + KEYSPACE
+ + "."
+ + TABLE_NAME_VARIABLE
+ + " ("
+ + TUPLE_ID_FIELD
+ + ", "
+ + TUPLE_COUNTER_FIELD
+ + ", "
+ + TUPLE_BATCHID_FIELD
+ + ") VALUES (?, ?, ?)";
+ private static final String SELECT_DATA_QUERY =
+ "SELECT * FROM " + KEYSPACE + "." + TABLE_NAME_VARIABLE + ';';
+
+ private static final Random random = new Random();
+ private int tableID;
+
+ private static final ArrayList<Tuple3<String, Integer, Integer>> collection =
+ new ArrayList<>(20);
+ private static final ArrayList<Row> rowCollection = new ArrayList<>(20);
+
+ private static final TypeInformation[] FIELD_TYPES = {
+ BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO
+ };
+
+ static {
+ for (int i = 0; i < 20; i++) {
+ collection.add(new Tuple3<>(UUID.randomUUID().toString(), i, 0));
+ rowCollection.add(Row.of(UUID.randomUUID().toString(), i, 0));
+ }
+ }
+
+ private static Class<? extends Pojo> annotatePojoWithTable(String keyspace, String tableName) {
+ return new ByteBuddy()
+ .redefine(Pojo.class)
+ .name("org.apache.flink.streaming.connectors.cassandra.Pojo" + tableName)
+ .annotateType(createTableAnnotation(keyspace, tableName))
+ .make()
+ .load(Pojo.class.getClassLoader())
+ .getLoaded();
+ }
+
+ @NotNull
+ private static Table createTableAnnotation(String keyspace, String tableName) {
+ return new Table() {
+
+ @Override
+ public String keyspace() {
+ return keyspace;
+ }
+
+ @Override
+ public String name() {
+ return tableName;
+ }
+
+ @Override
+ public boolean caseSensitiveKeyspace() {
+ return false;
+ }
+
+ @Override
+ public boolean caseSensitiveTable() {
+ return false;
+ }
+
+ @Override
+ public String writeConsistency() {
+ return "";
+ }
+
+ @Override
+ public String readConsistency() {
+ return "";
+ }
+
+ @Override
+ public Class<? extends Annotation> annotationType() {
+ return Table.class;
+ }
+ };
+ }
+
+ // ------------------------------------------------------------------------
+ // Utility methods
+ // ------------------------------------------------------------------------
+
+ public static CassandraContainer createCassandraContainer() {
+ CassandraContainer cassandra = new CassandraContainer(DockerImageVersions.CASSANDRA_4_0);
Review Comment:
> We'd like to preserve the git history; can you integrate the `cas_rewrite` branch from this repo into your PR?
That makes total sense. I'll rebase on the branch. IMHO it would be interesting to add an item to your migration doc to inform contributors to keep history, that way you would not need to create the branch yourself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot commented on a diff in pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot commented on code in PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#discussion_r1028212357
##########
pom.xml:
##########
@@ -0,0 +1,432 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>io.github.zentol.flink</groupId>
+ <artifactId>flink-connector-parent</artifactId>
+ <version>1.0</version>
+ </parent>
+
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-cassandra_${scala.binary.version}</artifactId>
+ <version>4.0-SNAPSHOT</version>
Review Comment:
:+1:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot commented on pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot commented on PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#issuecomment-1322108960
Thanks for the very prompt review @zentol !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot commented on a diff in pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot commented on code in PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#discussion_r1028076179
##########
src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java:
##########
@@ -0,0 +1,912 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.cassandra;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo;
+import org.apache.flink.batch.connectors.cassandra.CassandraInputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraPojoInputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraPojoOutputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraRowOutputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraTupleOutputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
+import org.apache.flink.streaming.runtime.operators.WriteAheadSinkTestBase;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.testutils.junit.RetryOnException;
+import org.apache.flink.testutils.junit.extensions.retry.RetryExtension;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.DockerImageVersions;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SimpleStatement;
+import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.Statement;
+import com.datastax.driver.core.exceptions.NoHostAvailableException;
+import com.datastax.driver.mapping.Mapper;
+import com.datastax.driver.mapping.annotations.Table;
+import net.bytebuddy.ByteBuddy;
+import org.assertj.core.api.recursive.comparison.RecursiveComparisonConfiguration;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.CassandraContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.io.IOException;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Constructor;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+
+import scala.collection.JavaConverters;
+import scala.collection.Seq;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for all cassandra sinks. */
+@SuppressWarnings("serial")
+// NoHostAvailableException is raised by Cassandra client under load while connecting to the cluster
+@RetryOnException(times = 3, exception = NoHostAvailableException.class)
+@Testcontainers
+@ExtendWith(RetryExtension.class)
+class CassandraConnectorITCase
+ extends WriteAheadSinkTestBase<
+ Tuple3<String, Integer, Integer>,
+ CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>>> {
+
+ private static final int MAX_CONNECTION_RETRY = 3;
+ private static final long CONNECTION_RETRY_DELAY = 500L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(CassandraConnectorITCase.class);
+ private static final Slf4jLogConsumer LOG_CONSUMER = new Slf4jLogConsumer(LOG);
+
+ @TempDir static Path tmpDir;
+
+ private static final int READ_TIMEOUT_MILLIS = 36000;
+
+ @Container static final CassandraContainer CASSANDRA_CONTAINER = createCassandraContainer();
+
+ private static final int PORT = 9042;
+
+ private static Cluster cluster;
+ private static Session session;
+
+ private final ClusterBuilder builderForReading =
+ createBuilderWithConsistencyLevel(ConsistencyLevel.ONE);
+ // Lower consistency level ANY is only available for writing.
+ private final ClusterBuilder builderForWriting =
+ createBuilderWithConsistencyLevel(ConsistencyLevel.ANY);
+
+ private ClusterBuilder createBuilderWithConsistencyLevel(ConsistencyLevel consistencyLevel) {
+ return new ClusterBuilder() {
+ @Override
+ protected Cluster buildCluster(Cluster.Builder builder) {
+ return builder.addContactPointsWithPorts(
+ new InetSocketAddress(
+ CASSANDRA_CONTAINER.getHost(),
+ CASSANDRA_CONTAINER.getMappedPort(PORT)))
+ .withQueryOptions(
+ new QueryOptions()
+ .setConsistencyLevel(consistencyLevel)
+ .setSerialConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL))
+ .withSocketOptions(
+ new SocketOptions()
+ // default timeout x 3
+ .setConnectTimeoutMillis(15000)
+ // default timeout x3 and higher than
+ // request_timeout_in_ms at the cluster level
+ .setReadTimeoutMillis(READ_TIMEOUT_MILLIS))
+ .withoutJMXReporting()
+ .withoutMetrics()
+ .build();
+ }
+ };
+ }
+
+ private static final String TABLE_NAME_PREFIX = "flink_";
+ private static final String TABLE_NAME_VARIABLE = "$TABLE";
+ private static final String KEYSPACE = "flink";
+ private static final String TUPLE_ID_FIELD = "id";
+ private static final String TUPLE_COUNTER_FIELD = "counter";
+ private static final String TUPLE_BATCHID_FIELD = "batch_id";
+ private static final String CREATE_KEYSPACE_QUERY =
+ "CREATE KEYSPACE "
+ + KEYSPACE
+ + " WITH replication= {'class':'SimpleStrategy', 'replication_factor':1};";
+ private static final String CREATE_TABLE_QUERY =
+ "CREATE TABLE "
+ + KEYSPACE
+ + "."
+ + TABLE_NAME_VARIABLE
+ + " ("
+ + TUPLE_ID_FIELD
+ + " text PRIMARY KEY, "
+ + TUPLE_COUNTER_FIELD
+ + " int, "
+ + TUPLE_BATCHID_FIELD
+ + " int);";
+ private static final String INSERT_DATA_QUERY =
+ "INSERT INTO "
+ + KEYSPACE
+ + "."
+ + TABLE_NAME_VARIABLE
+ + " ("
+ + TUPLE_ID_FIELD
+ + ", "
+ + TUPLE_COUNTER_FIELD
+ + ", "
+ + TUPLE_BATCHID_FIELD
+ + ") VALUES (?, ?, ?)";
+ private static final String SELECT_DATA_QUERY =
+ "SELECT * FROM " + KEYSPACE + "." + TABLE_NAME_VARIABLE + ';';
+
+ private static final Random random = new Random();
+ private int tableID;
+
+ private static final ArrayList<Tuple3<String, Integer, Integer>> collection =
+ new ArrayList<>(20);
+ private static final ArrayList<Row> rowCollection = new ArrayList<>(20);
+
+ private static final TypeInformation[] FIELD_TYPES = {
+ BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO
+ };
+
+ static {
+ for (int i = 0; i < 20; i++) {
+ collection.add(new Tuple3<>(UUID.randomUUID().toString(), i, 0));
+ rowCollection.add(Row.of(UUID.randomUUID().toString(), i, 0));
+ }
+ }
+
+ private static Class<? extends Pojo> annotatePojoWithTable(String keyspace, String tableName) {
+ return new ByteBuddy()
+ .redefine(Pojo.class)
+ .name("org.apache.flink.streaming.connectors.cassandra.Pojo" + tableName)
+ .annotateType(createTableAnnotation(keyspace, tableName))
+ .make()
+ .load(Pojo.class.getClassLoader())
+ .getLoaded();
+ }
+
+ @NotNull
+ private static Table createTableAnnotation(String keyspace, String tableName) {
+ return new Table() {
+
+ @Override
+ public String keyspace() {
+ return keyspace;
+ }
+
+ @Override
+ public String name() {
+ return tableName;
+ }
+
+ @Override
+ public boolean caseSensitiveKeyspace() {
+ return false;
+ }
+
+ @Override
+ public boolean caseSensitiveTable() {
+ return false;
+ }
+
+ @Override
+ public String writeConsistency() {
+ return "";
+ }
+
+ @Override
+ public String readConsistency() {
+ return "";
+ }
+
+ @Override
+ public Class<? extends Annotation> annotationType() {
+ return Table.class;
+ }
+ };
+ }
+
+ // ------------------------------------------------------------------------
+ // Utility methods
+ // ------------------------------------------------------------------------
+
+ public static CassandraContainer createCassandraContainer() {
+ CassandraContainer cassandra = new CassandraContainer(DockerImageVersions.CASSANDRA_4_0);
Review Comment:
> We'd like to preserve the git history; can you integrate the `cas_rewrite` branch from this repo into your PR?
That makes total sense. I'll rebase on the branch. IMHO it would be interesting to add an item to your migration doc to inform contributors to keep history, that way you would not need to create the branch yourself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot commented on pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot commented on PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#issuecomment-1322351036
> > > We'd like to preserve the git history; can you integrate the `cas_rewrite` branch from this repo into your PR?
> >
> >
> > That makes total sense. I'll rebase on the branch. IMHO it would be interesting to add an item to your migration doc to inform contributors to keep history, that way you would not need to create the branch yourself.
>
> @zentol: I rebased on cas_rerewrite and when I pushed to my fork it has closed the PR automatically with no action from my side. I don't have the rights to re-open it. Could you reopen this PR so that we don't lose your review comments?
>
> I messed up the rebase. I needed to rebase the other way around
I've fixed my rebase, do you prefer that you reopen this PR and I force push here or that I open another PR ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot closed pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot closed pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
URL: https://github.com/apache/flink-connector-cassandra/pull/1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] zentol commented on pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#issuecomment-1321863407
> We'd like to preserve the git history; can you integrate the cas_rewrite branch from this repo into your PR?
hold on I may want to update that branch to also contain the docs...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot commented on pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot commented on PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#issuecomment-1322134409
> > We'd like to preserve the git history; can you integrate the `cas_rewrite` branch from this repo into your PR?
>
> That makes total sense. I'll rebase on the branch. IMHO it would be interesting to add an item to your migration doc to inform contributors to keep history, that way you would not need to create the branch yourself.
@zentol:
I rebased on cas_rerewrite and when I pushed to my fork it has closed the PR automatically with no action from my side. I don't have the rights to re-open it. Could you reopen this PR so that we don't lose your review comments?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot commented on pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot commented on PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#issuecomment-1322126546
> We'd like to preserve the git history; can you integrate the `cas_rewrite` branch from this repo into your PR?
That makes total sense. I'll rebase on the branch. IMHO it would be interesting to add an item to your migration doc to inform contributors to keep history, that way you would not need to create the branch yourself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot commented on pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot commented on PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#issuecomment-1323381672
> > > > We'd like to preserve the git history; can you integrate the `cas_rewrite` branch from this repo into your PR?
> > >
> > >
> > > That makes total sense. I'll rebase on the branch. IMHO it would be interesting to add an item to your migration doc to inform contributors to keep history, that way you would not need to create the branch yourself.
> >
> >
> > @zentol: I rebased on cas_rerewrite and when I pushed to my fork it has closed the PR automatically with no action from my side. I don't have the rights to re-open it. Could you reopen this PR so that we don't lose your review comments?
> > I messed up the rebase. I needed to rebase the other way around
>
> @zentol I've fixed my rebase, do you prefer that you reopen this PR and I force push here or that I open another PR ?
I opened a new one: https://github.com/apache/flink-connector-cassandra/pull/2
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] echauchot commented on a diff in pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
echauchot commented on code in PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#discussion_r1028282565
##########
src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java:
##########
@@ -0,0 +1,912 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.cassandra;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo;
+import org.apache.flink.batch.connectors.cassandra.CassandraInputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraPojoInputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraPojoOutputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraRowOutputFormat;
+import org.apache.flink.batch.connectors.cassandra.CassandraTupleOutputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
+import org.apache.flink.streaming.runtime.operators.WriteAheadSinkTestBase;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.testutils.junit.RetryOnException;
+import org.apache.flink.testutils.junit.extensions.retry.RetryExtension;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.DockerImageVersions;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SimpleStatement;
+import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.Statement;
+import com.datastax.driver.core.exceptions.NoHostAvailableException;
+import com.datastax.driver.mapping.Mapper;
+import com.datastax.driver.mapping.annotations.Table;
+import net.bytebuddy.ByteBuddy;
+import org.assertj.core.api.recursive.comparison.RecursiveComparisonConfiguration;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.CassandraContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.io.IOException;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Constructor;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+
+import scala.collection.JavaConverters;
+import scala.collection.Seq;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for all cassandra sinks. */
+@SuppressWarnings("serial")
+// NoHostAvailableException is raised by Cassandra client under load while connecting to the cluster
+@RetryOnException(times = 3, exception = NoHostAvailableException.class)
+@Testcontainers
+@ExtendWith(RetryExtension.class)
+class CassandraConnectorITCase
+ extends WriteAheadSinkTestBase<
+ Tuple3<String, Integer, Integer>,
+ CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>>> {
+
+ private static final int MAX_CONNECTION_RETRY = 3;
+ private static final long CONNECTION_RETRY_DELAY = 500L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(CassandraConnectorITCase.class);
+ private static final Slf4jLogConsumer LOG_CONSUMER = new Slf4jLogConsumer(LOG);
+
+ @TempDir static Path tmpDir;
+
+ private static final int READ_TIMEOUT_MILLIS = 36000;
+
+ @Container static final CassandraContainer CASSANDRA_CONTAINER = createCassandraContainer();
+
+ private static final int PORT = 9042;
+
+ private static Cluster cluster;
+ private static Session session;
+
+ private final ClusterBuilder builderForReading =
+ createBuilderWithConsistencyLevel(ConsistencyLevel.ONE);
+ // Lower consistency level ANY is only available for writing.
+ private final ClusterBuilder builderForWriting =
+ createBuilderWithConsistencyLevel(ConsistencyLevel.ANY);
+
+ private ClusterBuilder createBuilderWithConsistencyLevel(ConsistencyLevel consistencyLevel) {
+ return new ClusterBuilder() {
+ @Override
+ protected Cluster buildCluster(Cluster.Builder builder) {
+ return builder.addContactPointsWithPorts(
+ new InetSocketAddress(
+ CASSANDRA_CONTAINER.getHost(),
+ CASSANDRA_CONTAINER.getMappedPort(PORT)))
+ .withQueryOptions(
+ new QueryOptions()
+ .setConsistencyLevel(consistencyLevel)
+ .setSerialConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL))
+ .withSocketOptions(
+ new SocketOptions()
+ // default timeout x 3
+ .setConnectTimeoutMillis(15000)
+ // default timeout x3 and higher than
+ // request_timeout_in_ms at the cluster level
+ .setReadTimeoutMillis(READ_TIMEOUT_MILLIS))
+ .withoutJMXReporting()
+ .withoutMetrics()
+ .build();
+ }
+ };
+ }
+
+ private static final String TABLE_NAME_PREFIX = "flink_";
+ private static final String TABLE_NAME_VARIABLE = "$TABLE";
+ private static final String KEYSPACE = "flink";
+ private static final String TUPLE_ID_FIELD = "id";
+ private static final String TUPLE_COUNTER_FIELD = "counter";
+ private static final String TUPLE_BATCHID_FIELD = "batch_id";
+ private static final String CREATE_KEYSPACE_QUERY =
+ "CREATE KEYSPACE "
+ + KEYSPACE
+ + " WITH replication= {'class':'SimpleStrategy', 'replication_factor':1};";
+ private static final String CREATE_TABLE_QUERY =
+ "CREATE TABLE "
+ + KEYSPACE
+ + "."
+ + TABLE_NAME_VARIABLE
+ + " ("
+ + TUPLE_ID_FIELD
+ + " text PRIMARY KEY, "
+ + TUPLE_COUNTER_FIELD
+ + " int, "
+ + TUPLE_BATCHID_FIELD
+ + " int);";
+ private static final String INSERT_DATA_QUERY =
+ "INSERT INTO "
+ + KEYSPACE
+ + "."
+ + TABLE_NAME_VARIABLE
+ + " ("
+ + TUPLE_ID_FIELD
+ + ", "
+ + TUPLE_COUNTER_FIELD
+ + ", "
+ + TUPLE_BATCHID_FIELD
+ + ") VALUES (?, ?, ?)";
+ private static final String SELECT_DATA_QUERY =
+ "SELECT * FROM " + KEYSPACE + "." + TABLE_NAME_VARIABLE + ';';
+
+ private static final Random random = new Random();
+ private int tableID;
+
+ private static final ArrayList<Tuple3<String, Integer, Integer>> collection =
+ new ArrayList<>(20);
+ private static final ArrayList<Row> rowCollection = new ArrayList<>(20);
+
+ private static final TypeInformation[] FIELD_TYPES = {
+ BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO
+ };
+
+ static {
+ for (int i = 0; i < 20; i++) {
+ collection.add(new Tuple3<>(UUID.randomUUID().toString(), i, 0));
+ rowCollection.add(Row.of(UUID.randomUUID().toString(), i, 0));
+ }
+ }
+
+ private static Class<? extends Pojo> annotatePojoWithTable(String keyspace, String tableName) {
+ return new ByteBuddy()
+ .redefine(Pojo.class)
+ .name("org.apache.flink.streaming.connectors.cassandra.Pojo" + tableName)
+ .annotateType(createTableAnnotation(keyspace, tableName))
+ .make()
+ .load(Pojo.class.getClassLoader())
+ .getLoaded();
+ }
+
+ @NotNull
+ private static Table createTableAnnotation(String keyspace, String tableName) {
+ return new Table() {
+
+ @Override
+ public String keyspace() {
+ return keyspace;
+ }
+
+ @Override
+ public String name() {
+ return tableName;
+ }
+
+ @Override
+ public boolean caseSensitiveKeyspace() {
+ return false;
+ }
+
+ @Override
+ public boolean caseSensitiveTable() {
+ return false;
+ }
+
+ @Override
+ public String writeConsistency() {
+ return "";
+ }
+
+ @Override
+ public String readConsistency() {
+ return "";
+ }
+
+ @Override
+ public Class<? extends Annotation> annotationType() {
+ return Table.class;
+ }
+ };
+ }
+
+ // ------------------------------------------------------------------------
+ // Utility methods
+ // ------------------------------------------------------------------------
+
+ public static CassandraContainer createCassandraContainer() {
+ CassandraContainer cassandra = new CassandraContainer(DockerImageVersions.CASSANDRA_4_0);
Review Comment:
:+1:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-cassandra] zentol commented on pull request #1: [FLINK-29982] move the existing Cassandra connector to external repository
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #1:
URL: https://github.com/apache/flink-connector-cassandra/pull/1#issuecomment-1322130904
> IMHO it would be interesting to add an item to your migration doc to inform contributors to keep history
I've already added this as an item to https://cwiki.apache.org/confluence/display/FLINK/Externalized+Connector+development.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org