You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by "jberragan (via GitHub)" <gi...@apache.org> on 2025/02/10 19:11:38 UTC
[PR] Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
jberragan opened a new pull request, #99:
URL: https://github.com/apache/cassandra-analytics/pull/99
(no comment)
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2006001237
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcClient.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import o.a.c.sidecar.client.shaded.common.utils.HttpRange;
+import org.apache.cassandra.cdc.api.CommitLog;
+import org.apache.cassandra.cdc.stats.ICdcStats;
+import org.apache.cassandra.clients.Sidecar;
+import org.apache.cassandra.sidecar.client.SidecarClient;
+import org.apache.cassandra.sidecar.client.SidecarInstance;
+import org.apache.cassandra.sidecar.client.StreamBuffer;
+import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
+import org.apache.cassandra.spark.exceptions.TransportFailureException;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.spark.utils.streaming.StreamConsumer;
+
+public class SidecarCdcClient
Review Comment:
We have to convert the HTTP clients PoJo objects into the CDC internal equivalent, i.e. `SegmentInfo` -> `SidecarCdcCommitLogSegment`, and handle status codes like `404` gracefully.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2001372287
##########
cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/CdcBuilder.java:
##########
@@ -41,26 +41,26 @@
public class CdcBuilder
{
@NotNull
- final String jobId;
- final int partitionId;
+ protected final String jobId;
+ protected final int partitionId;
@NotNull
TokenRangeSupplier tokenRangeSupplier = () -> null;
@NotNull
- SchemaSupplier schemaSupplier;
+ protected SchemaSupplier schemaSupplier;
@NotNull
CassandraSource cassandraSource = CassandraSource.DEFAULT;
@NotNull
StatePersister statePersister = StatePersister.STUB;
@NotNull
- CdcOptions cdcOptions = CdcOptions.DEFAULT;
+ protected CdcOptions cdcOptions = CdcOptions.DEFAULT;
@NotNull
ICdcStats stats = ICdcStats.STUB;
@Nullable
Review Comment:
At compile time it is `Nullable`, at runtime it is illegal to not provide. The issue is I don't have a ExecutorSerivce around to provide a default `AsyncExecutor` and I don't want to create one just for the default.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2001528452
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarStatePersister.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.ThreadLocalMonotonicTimestampGenerator;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
+import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.cdc.CdcKryoRegister;
+import org.apache.cassandra.cdc.api.CdcOptions;
+import org.apache.cassandra.cdc.api.StatePersister;
+import org.apache.cassandra.cdc.state.CdcState;
+import org.apache.cassandra.spark.utils.AsyncExecutor;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.util.CompressionUtil;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * SidecarStatePersister buffers CDC state and flushes at regular time intervals, so we only write the latest CDC state and don't wastefully write expired data.
+ */
+public class SidecarStatePersister implements StatePersister
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(SidecarStatePersister.class);
+
+ // group latest state by jobId/token range, so we persist independently
+ protected final ConcurrentHashMap<PersistWrapper.Key, PersistWrapper> latestState = new ConcurrentHashMap<>();
+ protected final ConcurrentLinkedQueue<TimedFutureWrapper> activeFlush = new ConcurrentLinkedQueue<>();
+ private final ThreadLocalMonotonicTimestampGenerator timestampGenerator = new ThreadLocalMonotonicTimestampGenerator();
+ private final SidecarCdcOptions sidecarCdcOptions;
+ private final CdcOptions cdcOptions;
+ private final SidecarCdcCassandraClient cassandraClient;
+ private final SidecarCdcStats sidecarCdcStats;
+ private final AsyncExecutor asyncExecutor;
+ volatile long timerId = -1L;
+
+ public SidecarStatePersister(SidecarCdcOptions sidecarCdcOptions,
+ CdcOptions cdcOptions,
+ SidecarCdcStats sidecarCdcStats,
+ SidecarCdcCassandraClient cassandraClient,
+ AsyncExecutor asyncExecutor)
+ {
+ this.sidecarCdcOptions = sidecarCdcOptions;
+ this.cdcOptions = cdcOptions;
+ this.sidecarCdcStats = sidecarCdcStats;
+ this.cassandraClient = cassandraClient;
+ this.asyncExecutor = asyncExecutor;
+ }
+
+ // StatePersister implemented methods
+
+ @Override
+ public void persist(String jobId, int partitionId, @Nullable TokenRange tokenRange, @NotNull ByteBuffer buf)
+ {
+ final PersistWrapper latest = new PersistWrapper(jobId, partitionId, tokenRange, buf, timestampGenerator.next());
+ PersistWrapper.Key key = latest.key();
+ if (!latest.equals(this.latestState.get(key)))
+ {
+ this.latestState.compute(key, (k, prev) -> !latest.equals(prev) ? latest : prev);
+ }
+ }
+
+ @NotNull
+ @Override
+ public List<CdcState> loadState(String jobId, int partitionId, @Nullable TokenRange tokenRange)
+ {
+ CompressionUtil compressionUtil = CdcBridgeFactory.get(cdcOptions.version()).compressionUtil();
+ List<Integer> sizes = new ArrayList<>();
+ // deserialize and merge the CDC state objects into canonical view
+ List<CdcState> result = loadStateForRange(jobId, tokenRange)
+ .peek(bytes -> sizes.add(bytes.length))
+ .map(bytes -> CdcState.deserialize(CdcKryoRegister.kryo(), compressionUtil, bytes))
Review Comment:
they are the same. I think it is less common to see the usage `peek` API. Meanwhile, the suggestion is more commonly used.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jyothsnakonisa (via GitHub)" <gi...@apache.org>.
jyothsnakonisa commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2002031033
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcClient.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import o.a.c.sidecar.client.shaded.common.utils.HttpRange;
+import org.apache.cassandra.cdc.api.CommitLog;
+import org.apache.cassandra.cdc.stats.ICdcStats;
+import org.apache.cassandra.clients.Sidecar;
+import org.apache.cassandra.sidecar.client.SidecarClient;
+import org.apache.cassandra.sidecar.client.SidecarInstance;
+import org.apache.cassandra.sidecar.client.StreamBuffer;
+import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
+import org.apache.cassandra.spark.exceptions.TransportFailureException;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.spark.utils.streaming.StreamConsumer;
+
+public class SidecarCdcClient
Review Comment:
Do we need a separate `SidecarCdcClient` class? Now that `streamCdcSegments` and `listCdcSegments` are added to SidecarClient, may be this class is no longer needed. All this class is doing Is making delegate call to above methods and processing the response may be we can move that processing response logic.
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcOptions.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+public interface SidecarCdcOptions
Review Comment:
We have this interface in sidecar, https://github.com/apache/cassandra-sidecar/blob/trunk/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcConfig.java
We can import the one from sidecar and get rid of this interface
##########
scripts/build-sidecar.sh:
##########
@@ -44,7 +44,7 @@ if [ "x${LOCAL_SIDECAR_REPO}" != "x" ]; then
else
SIDECAR_REPO="${SIDECAR_REPO:-https://github.com/apache/cassandra-sidecar.git}"
SIDECAR_BRANCH="${SIDECAR_BRANCH:-trunk}"
- SIDECAR_COMMIT="${SIDECAR_COMMIT:-f3bcbba3dcd81b640711baa35a76a2d949ce6c5e}"
+ SIDECAR_COMMIT="${SIDECAR_COMMIT:-c1260886642e789fa15a3fb900251bea529fcc09}"
Review Comment:
Since we have OSS sidecar version release, can we use that instead?
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcCommitLogSegment.java:
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.time.Duration;
+
+import o.a.c.sidecar.client.shaded.common.response.data.CdcSegmentInfo;
+import o.a.c.sidecar.client.shaded.common.utils.HttpRange;
+import org.apache.cassandra.cdc.api.CommitLog;
+import org.apache.cassandra.cdc.stats.ICdcStats;
+import org.apache.cassandra.clients.Sidecar;
+import org.apache.cassandra.spark.data.FileType;
+import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
+import org.apache.cassandra.spark.utils.streaming.CassandraFileSource;
+import org.apache.cassandra.spark.utils.streaming.StreamConsumer;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * The SidecarCdcCommitLogSegment provides a CommitLog source for reading the
+ * CommitLog segments over the Sidecar HTTP API.
+ */
+public class SidecarCdcCommitLogSegment implements CommitLog
+{
+ private final CassandraInstance instance;
+ private final CdcSegmentInfo segment;
+ private final CassandraFileSource<CommitLog> source;
+ private final ICdcStats stats;
+
+ public SidecarCdcCommitLogSegment(SidecarCdcClient sidecar,
+ CassandraInstance instance,
+ CdcSegmentInfo segment,
+ Sidecar.ClientConfig clientConfig)
+ {
+ this.instance = instance;
+ this.segment = segment;
+ this.stats = sidecar.stats;
+
+ final SidecarCdcCommitLogSegment thisLog = this;
+ this.source = new CassandraFileSource<CommitLog>()
+ {
+ public void request(long start, long end, StreamConsumer consumer)
+ {
+ sidecar.streamCdcCommitLogSegment(instance, segment.name, HttpRange.of(start, end), consumer);
+ }
+
+ public CommitLog cassandraFile()
+ {
+ return thisLog;
+ }
+
+ public long maxBufferSize()
+ {
+ return clientConfig.maxBufferSize();
+ }
+
+ public long chunkBufferSize()
+ {
+ return clientConfig.chunkBufferSize(fileType());
+ }
+
+ public FileType fileType()
+ {
+ return FileType.COMMITLOG;
+ }
+
+ public long size()
+ {
+ return segment.idx;
+ }
+
+ @Nullable
+ @Override
+ public Duration timeout()
+ {
+ final int timeout = clientConfig.timeoutSeconds();
Review Comment:
we have not been using final for local variables, can we please remove final here and in other places for the sake of consistency.
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarStatePersister.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.ThreadLocalMonotonicTimestampGenerator;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
+import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.cdc.CdcKryoRegister;
+import org.apache.cassandra.cdc.api.CdcOptions;
+import org.apache.cassandra.cdc.api.StatePersister;
+import org.apache.cassandra.cdc.state.CdcState;
+import org.apache.cassandra.spark.utils.AsyncExecutor;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.util.CompressionUtil;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * SidecarStatePersister buffers CDC state and flushes at regular time intervals, so we only write the latest CDC state and don't wastefully write expired data.
+ */
+public class SidecarStatePersister implements StatePersister
Review Comment:
I see a StatePersister in sidecar but the implementation is is a little different between two classes, can we potentially get rid of the one in Sidecar?
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2001525702
##########
cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/CdcBuilder.java:
##########
@@ -41,26 +41,26 @@
public class CdcBuilder
{
@NotNull
- final String jobId;
- final int partitionId;
+ protected final String jobId;
+ protected final int partitionId;
@NotNull
TokenRangeSupplier tokenRangeSupplier = () -> null;
@NotNull
- SchemaSupplier schemaSupplier;
+ protected SchemaSupplier schemaSupplier;
@NotNull
CassandraSource cassandraSource = CassandraSource.DEFAULT;
@NotNull
StatePersister statePersister = StatePersister.STUB;
@NotNull
- CdcOptions cdcOptions = CdcOptions.DEFAULT;
+ protected CdcOptions cdcOptions = CdcOptions.DEFAULT;
@NotNull
ICdcStats stats = ICdcStats.STUB;
@Nullable
Review Comment:
I find it confusing to see the annotation is not reflecting the runtime expectation. IMO, it is fine to ignore the warning at the `@NotNull` annotation. Or, if we want to mute the warning, the initialization can be moved to the constructor.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2001378436
##########
cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java:
##########
@@ -122,51 +119,6 @@ public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvid
return buildClient(sidecarConfig, vertx, httpClientConfig, sidecarInstancesProvider);
}
- static String transportModeBasedWriterUserAgent(DataTransport transport)
- {
- switch (transport)
- {
- case S3_COMPAT:
- return BuildInfo.WRITER_S3_USER_AGENT;
- case DIRECT:
- default:
- return BuildInfo.WRITER_USER_AGENT;
- }
- }
-
- public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvider, BulkSparkConf conf)
- {
- Vertx vertx = Vertx.vertx(new VertxOptions().setUseDaemonThread(true)
- .setWorkerPoolSize(conf.getMaxHttpConnections()));
-
Review Comment:
The `BulkSparkConf` cruft for the bulk writer gets in the way.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "bbotella (via GitHub)" <gi...@apache.org>.
bbotella merged PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#issuecomment-2707911120
Latest green CI: https://app.circleci.com/pipelines/github/jberragan/cassandra-analytics/180/workflows/867d6289-4c5e-4658-8811-b43f91fa7753
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#issuecomment-2657787970
Latest CircleCI: https://app.circleci.com/pipelines/github/jberragan/cassandra-analytics/170/workflows/05154e78-9211-429b-b290-720cfe6eb090
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2003790383
##########
scripts/build-sidecar.sh:
##########
@@ -44,7 +44,7 @@ if [ "x${LOCAL_SIDECAR_REPO}" != "x" ]; then
else
SIDECAR_REPO="${SIDECAR_REPO:-https://github.com/apache/cassandra-sidecar.git}"
SIDECAR_BRANCH="${SIDECAR_BRANCH:-trunk}"
- SIDECAR_COMMIT="${SIDECAR_COMMIT:-f3bcbba3dcd81b640711baa35a76a2d949ce6c5e}"
+ SIDECAR_COMMIT="${SIDECAR_COMMIT:-c1260886642e789fa15a3fb900251bea529fcc09}"
Review Comment:
👍 trunk is now using the release 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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r1992355072
##########
cassandra-analytics-cdc-sidecar/build.gradle:
##########
@@ -0,0 +1,142 @@
+import java.nio.file.Paths
+
+/*
+ * 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.
+ */
+
+plugins {
+ id('java-library')
+ id('maven-publish')
+}
+
+configurations {
+ all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
+ all*.exclude(group: 'log4j', module: 'log4j')
+}
+
+java {
+ withJavadocJar()
+ withSourcesJar()
+}
+
+publishing {
+ publications {
+ maven(MavenPublication) {
+ from components.java
+ groupId project.group
+ artifactId "${archivesBaseName}"
+ version System.getenv("CODE_VERSION") ?: "${version}"
+ }
+ }
+}
+
+configurations {
+ fourzero {
+ description = 'Cassandra 4.0 dependency'
+ }
+ fourzerobridge {
+ description = 'Bridge to Cassandra 4.0'
+ }
+ fourzerotypes {
+ description = 'Cassandra 4.0 Data Types'
+ }
+}
+
+dependencies {
+ api(project(':cassandra-bridge'))
+
+ fourzero(project(path: ':cassandra-four-zero', configuration: 'shadow'))
+ fourzerobridge(project(path: ':cassandra-four-zero-bridge'))
+ fourzerotypes(project(path: ':cassandra-four-zero-types'))
+
+ implementation project(":cassandra-analytics-common")
+ implementation project(':cassandra-analytics-cdc')
+ implementation project(':cassandra-analytics-sidecar-client')
+ compileOnlyApi("com.datastax.cassandra:cassandra-driver-core:${project.cassandraDriverVersion}") // Cassandra driver is required only if user supplies the CassandraClientSource
+
+ implementation "com.google.guava:guava:${project.guavaVersion}"
+
+ implementation "org.apache.commons:commons-lang3:${project.commonsLang3Version}"
+ implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
+ implementation "com.esotericsoftware:kryo-shaded:${kryoVersion}"
+ implementation "com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}"
+ implementation "com.fasterxml.jackson.core:jackson-core:${jacksonVersion}"
+ implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
+
+ testImplementation project(":cassandra-analytics-common")
+
+ // pull in cassandra-bridge so we can re-use TestSchema to generate arbitrary schemas for the cdc tests
+ testImplementation project(":cassandra-bridge")
+ testImplementation("com.datastax.cassandra:cassandra-driver-core:${project.cassandraDriverVersion}")
+ testImplementation(testFixtures(project(':cassandra-bridge')))
+ testImplementation project(":cassandra-four-zero-bridge")
+ testImplementation project(":cassandra-four-zero-types")
+ testImplementation project(path: ':cassandra-four-zero', configuration: 'shadow')
+ testImplementation(group: 'org.quicktheories', name: 'quicktheories', version: "${project.rootProject.quickTheoriesVersion}")
+ testImplementation(group: 'com.google.guava', name: 'guava', version: '31.1-jre')
Review Comment:
I had to revert back to `31.1-jre`, for tests it needs to match the Cassandra guava version otherwise it results in:
```
java.lang.NoSuchMethodError: 'void com.google.common.base.Preconditions.checkArgument(boolean, java.lang.String, java.lang.Object)'
```
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#issuecomment-2652272160
Green CI: https://app.circleci.com/pipelines/github/jberragan/cassandra-analytics/164/workflows/e8b6d7f5-2119-4abf-af93-2c81cb9ffd36
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "bbotella (via GitHub)" <gi...@apache.org>.
bbotella commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r1980284019
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcBuilder.java:
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.io.IOException;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.cassandra.cdc.CdcBuilder;
+import org.apache.cassandra.cdc.api.CdcOptions;
+import org.apache.cassandra.cdc.api.EventConsumer;
+import org.apache.cassandra.cdc.api.SchemaSupplier;
+import org.apache.cassandra.cdc.api.TokenRangeSupplier;
+import org.apache.cassandra.cdc.stats.ICdcStats;
+import org.apache.cassandra.clients.Sidecar;
+import org.apache.cassandra.secrets.SecretsProvider;
+import org.apache.cassandra.sidecar.client.SidecarClient;
+import org.apache.cassandra.sidecar.client.SidecarInstancesProvider;
+import org.apache.cassandra.spark.utils.AsyncExecutor;
+import org.jetbrains.annotations.NotNull;
+
+@SuppressWarnings("unused")
+public class SidecarCdcBuilder extends CdcBuilder
+{
+ protected ClusterConfigProvider clusterConfigProvider;
+ protected SidecarCdcClient sidecarCdcClient;
+ protected SidecarDownMonitor downMonitor = SidecarDownMonitor.STUB;
+ protected ReplicationFactorSupplier replicationFactorSupplier = ReplicationFactorSupplier.DEFAULT;
+ protected SidecarCdcStats sidecarCdcStats = SidecarCdcStats.STUB;
+ protected SidecarCdcCassandraClient cassandraClient = SidecarCdcCassandraClient.STUB;
+ protected SidecarCdcOptions sidecarCdcOptions = SidecarCdcOptions.DEFAULT;
+
+ SidecarCdcBuilder(@NotNull String jobId,
+ int partitionId,
+ CdcOptions cdcOptions,
+ ClusterConfigProvider clusterConfigProvider,
+ EventConsumer eventConsumer,
+ SchemaSupplier schemaSupplier,
+ TokenRangeSupplier tokenRangeSupplier,
+ SidecarInstancesProvider sidecarInstancesProvider,
+ Sidecar.ClientConfig clientConfig,
+ SecretsProvider secretsProvider,
+ ICdcStats cdcStats) throws IOException
+ {
+ this(
+ jobId,
+ partitionId,
+ cdcOptions,
+ clusterConfigProvider,
+ eventConsumer,
+ schemaSupplier,
+ tokenRangeSupplier,
+ clientConfig,
+ Sidecar.from(sidecarInstancesProvider, clientConfig, secretsProvider),
+ cdcStats
+ );
+ }
+
+ SidecarCdcBuilder(@NotNull String jobId,
+ int partitionId,
+ CdcOptions cdcOptions,
+ ClusterConfigProvider clusterConfigProvider,
+ EventConsumer eventConsumer,
+ SchemaSupplier schemaSupplier,
+ TokenRangeSupplier tokenRangeSupplier,
+ Sidecar.ClientConfig clientConfig,
+ SidecarClient sidecarClient,
+ ICdcStats cdcStats)
+ {
+ super(jobId, partitionId, eventConsumer, schemaSupplier);
+ this.clusterConfigProvider = clusterConfigProvider;
+ this.sidecarCdcClient = new SidecarCdcClient(clientConfig, sidecarClient, cdcStats);
+ withCdcOptions(cdcOptions);
+ withTokenRangeSupplier(tokenRangeSupplier);
+ rebuildCommitLogProvider();
+ }
+
+ public SidecarCdcBuilder withClusterConfigProvider(ClusterConfigProvider clusterConfigProvider)
+ {
+ this.clusterConfigProvider = clusterConfigProvider;
+ rebuildCommitLogProvider();
+ return this;
+ }
+
+ public SidecarCdcBuilder withDownMonitor(SidecarDownMonitor downMonitor)
+ {
+ this.downMonitor = downMonitor;
+ rebuildCommitLogProvider();
+ return this;
+ }
+
+ public SidecarCdcBuilder withSidecarClient(Sidecar.ClientConfig clientConfig,
+ SidecarClient sidecarClient,
+ ICdcStats cdcStats)
+ {
+ this.sidecarCdcClient = new SidecarCdcClient(clientConfig, sidecarClient, cdcStats);
+ rebuildCommitLogProvider();
+ return this;
+ }
+
+ public SidecarCdcBuilder withReplicationFactorSupplier(ReplicationFactorSupplier replicationFactorSupplier)
+ {
+ this.replicationFactorSupplier = replicationFactorSupplier;
+ rebuildCommitLogProvider();
Review Comment:
Why are we calling `rebuildCommitLogProvider` on most of `with` methods? Can't we just call this on the build phase?
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2003610171
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarStatePersister.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.ThreadLocalMonotonicTimestampGenerator;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
+import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.cdc.CdcKryoRegister;
+import org.apache.cassandra.cdc.api.CdcOptions;
+import org.apache.cassandra.cdc.api.StatePersister;
+import org.apache.cassandra.cdc.state.CdcState;
+import org.apache.cassandra.spark.utils.AsyncExecutor;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.util.CompressionUtil;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * SidecarStatePersister buffers CDC state and flushes at regular time intervals, so we only write the latest CDC state and don't wastefully write expired data.
+ */
+public class SidecarStatePersister implements StatePersister
Review Comment:
Yes, we no longer need the `SidecarStatePersister` in the Sidecar 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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2003615209
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcOptions.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+public interface SidecarCdcOptions
Review Comment:
I would like to but as much Sidecar-CDC code in this module.
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcOptions.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+public interface SidecarCdcOptions
Review Comment:
I would like to put as much Sidecar-CDC code in this module.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2001378436
##########
cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java:
##########
@@ -122,51 +119,6 @@ public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvid
return buildClient(sidecarConfig, vertx, httpClientConfig, sidecarInstancesProvider);
}
- static String transportModeBasedWriterUserAgent(DataTransport transport)
- {
- switch (transport)
- {
- case S3_COMPAT:
- return BuildInfo.WRITER_S3_USER_AGENT;
- case DIRECT:
- default:
- return BuildInfo.WRITER_USER_AGENT;
- }
- }
-
- public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvider, BulkSparkConf conf)
- {
- Vertx vertx = Vertx.vertx(new VertxOptions().setUseDaemonThread(true)
- .setWorkerPoolSize(conf.getMaxHttpConnections()));
-
Review Comment:
The `BulkSparkConf` cruft gets in the way.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#issuecomment-2725341405
Do you have any other comments @yifan-c ?
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r1996359837
##########
cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/CdcBuilder.java:
##########
@@ -41,26 +41,26 @@
public class CdcBuilder
{
@NotNull
- final String jobId;
- final int partitionId;
+ protected final String jobId;
+ protected final int partitionId;
@NotNull
TokenRangeSupplier tokenRangeSupplier = () -> null;
@NotNull
- SchemaSupplier schemaSupplier;
+ protected SchemaSupplier schemaSupplier;
@NotNull
CassandraSource cassandraSource = CassandraSource.DEFAULT;
@NotNull
StatePersister statePersister = StatePersister.STUB;
@NotNull
- CdcOptions cdcOptions = CdcOptions.DEFAULT;
+ protected CdcOptions cdcOptions = CdcOptions.DEFAULT;
@NotNull
ICdcStats stats = ICdcStats.STUB;
@Nullable
Review Comment:
`asyncExecutor` is not `Nullable`. The builder cannot be built, if the field is `null`.
Update the annotation to `@NotNull`?
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarStatePersister.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.ThreadLocalMonotonicTimestampGenerator;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
+import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.cdc.CdcKryoRegister;
+import org.apache.cassandra.cdc.api.CdcOptions;
+import org.apache.cassandra.cdc.api.StatePersister;
+import org.apache.cassandra.cdc.state.CdcState;
+import org.apache.cassandra.spark.utils.AsyncExecutor;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.util.CompressionUtil;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * SidecarStatePersister buffers CDC state and flushes at regular time intervals, so we only write the latest CDC state and don't wastefully write expired data.
+ */
+public class SidecarStatePersister implements StatePersister
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(SidecarStatePersister.class);
+
+ // group latest state by jobId/token range, so we persist independently
+ protected final ConcurrentHashMap<PersistWrapper.Key, PersistWrapper> latestState = new ConcurrentHashMap<>();
+ protected final ConcurrentLinkedQueue<TimedFutureWrapper> activeFlush = new ConcurrentLinkedQueue<>();
+ private final ThreadLocalMonotonicTimestampGenerator timestampGenerator = new ThreadLocalMonotonicTimestampGenerator();
+ private final SidecarCdcOptions sidecarCdcOptions;
+ private final CdcOptions cdcOptions;
+ private final SidecarCdcCassandraClient cassandraClient;
+ private final SidecarCdcStats sidecarCdcStats;
+ private final AsyncExecutor asyncExecutor;
+ volatile long timerId = -1L;
+
+ public SidecarStatePersister(SidecarCdcOptions sidecarCdcOptions,
+ CdcOptions cdcOptions,
+ SidecarCdcStats sidecarCdcStats,
+ SidecarCdcCassandraClient cassandraClient,
+ AsyncExecutor asyncExecutor)
+ {
+ this.sidecarCdcOptions = sidecarCdcOptions;
+ this.cdcOptions = cdcOptions;
+ this.sidecarCdcStats = sidecarCdcStats;
+ this.cassandraClient = cassandraClient;
+ this.asyncExecutor = asyncExecutor;
+ }
+
+ // StatePersister implemented methods
+
+ @Override
+ public void persist(String jobId, int partitionId, @Nullable TokenRange tokenRange, @NotNull ByteBuffer buf)
+ {
+ final PersistWrapper latest = new PersistWrapper(jobId, partitionId, tokenRange, buf, timestampGenerator.next());
+ PersistWrapper.Key key = latest.key();
+ if (!latest.equals(this.latestState.get(key)))
+ {
+ this.latestState.compute(key, (k, prev) -> !latest.equals(prev) ? latest : prev);
+ }
+ }
+
+ @NotNull
+ @Override
+ public List<CdcState> loadState(String jobId, int partitionId, @Nullable TokenRange tokenRange)
+ {
+ CompressionUtil compressionUtil = CdcBridgeFactory.get(cdcOptions.version()).compressionUtil();
+ List<Integer> sizes = new ArrayList<>();
+ // deserialize and merge the CDC state objects into canonical view
+ List<CdcState> result = loadStateForRange(jobId, tokenRange)
+ .peek(bytes -> sizes.add(bytes.length))
+ .map(bytes -> CdcState.deserialize(CdcKryoRegister.kryo(), compressionUtil, bytes))
+ .collect(Collectors.toList());
+ int count = sizes.size();
+ int len = sizes.stream().mapToInt(i -> i).sum();
+ LOGGER.debug("Read CDC state from Cassandra jobId={} start={} end={} stateCount={} stateSize={}",
+ jobId, tokenRange == null ? "null" : tokenRange.lowerEndpoint(), tokenRange == null ? "null" : tokenRange.upperEndpoint(), count, len);
+ sidecarCdcStats.captureCdcConsumerReadFromState(count, len);
+ return result;
+ }
+
+ @VisibleForTesting
+ public Stream<byte[]> loadStateForRange(String jobId, @Nullable TokenRange tokenRange)
+ {
+ return cassandraClient
+ .loadStateForRange(jobId, tokenRange);
+ }
+
+ /**
+ * Start the SidecarStatePersister to flush to Cassandra every `persistDelay()`
+ */
+ public synchronized void start()
+ {
+ if (timerId >= 0)
+ {
+ // already running
+ return;
+ }
+ this.timerId = asyncExecutor.periodicTimer(this::persistToCassandra, sidecarCdcOptions.persistDelay().toMillis());
+ }
+
+ /**
+ * Stop the SidecarStatePersister gracefully, blocking to await for any pending flushes to complete.
+ */
+ public void stop()
+ {
+ stop(true);
+ }
+
+ public synchronized void stop(boolean flush)
+ {
+ if (this.timerId < 0)
+ {
+ // not running
+ return;
+ }
+
+ asyncExecutor.cancelTimer(this.timerId);
+ this.timerId = -1;
+
+ if (flush)
+ {
+ flush();
+ }
+ }
+
+ // internal methods
+
+ protected void persistToCassandra()
+ {
+ persistToCassandra(false);
+ }
+
+ protected void persistToCassandra(boolean force)
+ {
+ // clean-up finished futures
+ activeFlush.removeIf(wrapper -> {
+ if (wrapper.allDone())
+ {
+ try
+ {
+ wrapper.await();
+ sidecarCdcStats.capturePersistSucceeded(System.nanoTime() - wrapper.startTimeNanos);
+ }
+ catch (InterruptedException e)
+ {
+ LOGGER.warn("Persist failed with InterruptedException", e);
+ Thread.currentThread().interrupt();
+ sidecarCdcStats.capturePersistFailed(e);
+ }
+ catch (ExecutionException e)
+ {
+ LOGGER.warn("Persist failed", e);
+ sidecarCdcStats.capturePersistFailed(e);
+ }
Review Comment:
Maybe just catch `Throwable` to be safe.
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcClient.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import o.a.c.sidecar.client.shaded.common.utils.HttpRange;
+import org.apache.cassandra.cdc.api.CommitLog;
+import org.apache.cassandra.cdc.stats.ICdcStats;
+import org.apache.cassandra.clients.Sidecar;
+import org.apache.cassandra.secrets.SecretsProvider;
+import org.apache.cassandra.sidecar.client.SidecarClient;
+import org.apache.cassandra.sidecar.client.SidecarInstance;
+import org.apache.cassandra.sidecar.client.SidecarInstancesProvider;
+import org.apache.cassandra.sidecar.client.StreamBuffer;
+import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
+import org.apache.cassandra.spark.exceptions.TransportFailureException;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.spark.utils.streaming.StreamConsumer;
+
+public class SidecarCdcClient
+{
+ final Sidecar.ClientConfig config;
+ final SidecarClient sidecarClient;
+ final ICdcStats stats;
+
+ public SidecarCdcClient(Sidecar.ClientConfig config,
+ SidecarClient sidecarClient,
+ ICdcStats stats)
+ {
+ this.config = config;
+ this.sidecarClient = sidecarClient;
+ this.stats = stats;
+ }
+
+ public CompletableFuture<List<CommitLog>> listCdcCommitLogSegments(CassandraInstance instance)
+ {
+ return sidecarClient.listCdcSegments(toSidecarInstance(instance))
+ .thenApply(
+ response ->
+ response.segmentInfos()
+ .stream()
+ .map(segment -> (CommitLog) new SidecarCdcCommitLogSegment(this, instance, segment, config))
+ .collect(Collectors.toList())
+ ).exceptionally(throwable -> {
+ final Throwable cause = ThrowableUtils.rootCause(throwable);
+ if (cause instanceof TransportFailureException.Nonretryable
+ && ((TransportFailureException.Nonretryable) cause).isNotFound())
+ {
+ // Rescue the 404 not found exception - it is a permitted error
+ return Collections.emptyList();
+ }
+ // Rethrow the other exception
+ if (throwable instanceof Error)
+ {
+ throw (Error) throwable;
+ }
+ throw new RuntimeException(cause);
+ });
+ }
+
+ public void streamCdcCommitLogSegment(CassandraInstance instance, String segment, HttpRange httpRange, StreamConsumer streamConsumer)
+ {
+ sidecarClient.streamCdcSegments(toSidecarInstance(instance), segment, httpRange, new org.apache.cassandra.sidecar.client.StreamConsumer()
+ {
+ @Override
+ public void onRead(StreamBuffer streamBuffer)
+ {
+ streamConsumer.onRead(new org.apache.cassandra.spark.utils.streaming.StreamBuffer()
+ {
+ @Override
+ public void getBytes(int index, ByteBuffer destination, int length)
+ {
+ streamBuffer.copyBytes(index, destination, length);
+ }
+
+ @Override
+ public void getBytes(int index, byte[] destination, int destinationIndex, int length)
+ {
+ streamBuffer.copyBytes(index, destination, destinationIndex, length);
+ }
+
+ @Override
+ public byte getByte(int index)
+ {
+ return streamBuffer.getByte(index);
+ }
+
+ @Override
+ public int readableBytes()
+ {
+ return streamBuffer.readableBytes();
+ }
+
+ @Override
+ public void release()
+ {
+ streamBuffer.release();
+ }
+ });
+ }
+
+ @Override
+ public void onComplete()
+ {
+ streamConsumer.onEnd();
+ }
+
+ @Override
+ public void onError(Throwable throwable)
+ {
+ streamConsumer.onError(throwable);
+ }
+ });
+ }
+
+ protected SidecarInstance toSidecarInstance(CassandraInstance instance)
+ {
+ return new SidecarInstance()
+ {
+ @Override
+ public int port()
+ {
+ return config.effectivePort();
+ }
+
+ @Override
+ public String hostname()
+ {
+ return instance.nodeName();
+ }
+ };
+ }
+
+ public static SidecarCdcClient from(SidecarInstancesProvider sidecarInstancesProvider,
+ Sidecar.ClientConfig config,
+ SecretsProvider secretsProvider,
+ ICdcStats stats) throws IOException
+ {
+ return new SidecarCdcClient(config, Sidecar.from(sidecarInstancesProvider, config, secretsProvider), stats);
+ }
Review Comment:
Unused.
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarStatePersister.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.ThreadLocalMonotonicTimestampGenerator;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
+import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.cdc.CdcKryoRegister;
+import org.apache.cassandra.cdc.api.CdcOptions;
+import org.apache.cassandra.cdc.api.StatePersister;
+import org.apache.cassandra.cdc.state.CdcState;
+import org.apache.cassandra.spark.utils.AsyncExecutor;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.util.CompressionUtil;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * SidecarStatePersister buffers CDC state and flushes at regular time intervals, so we only write the latest CDC state and don't wastefully write expired data.
+ */
+public class SidecarStatePersister implements StatePersister
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(SidecarStatePersister.class);
+
+ // group latest state by jobId/token range, so we persist independently
+ protected final ConcurrentHashMap<PersistWrapper.Key, PersistWrapper> latestState = new ConcurrentHashMap<>();
+ protected final ConcurrentLinkedQueue<TimedFutureWrapper> activeFlush = new ConcurrentLinkedQueue<>();
+ private final ThreadLocalMonotonicTimestampGenerator timestampGenerator = new ThreadLocalMonotonicTimestampGenerator();
+ private final SidecarCdcOptions sidecarCdcOptions;
+ private final CdcOptions cdcOptions;
+ private final SidecarCdcCassandraClient cassandraClient;
+ private final SidecarCdcStats sidecarCdcStats;
+ private final AsyncExecutor asyncExecutor;
+ volatile long timerId = -1L;
+
+ public SidecarStatePersister(SidecarCdcOptions sidecarCdcOptions,
+ CdcOptions cdcOptions,
+ SidecarCdcStats sidecarCdcStats,
+ SidecarCdcCassandraClient cassandraClient,
+ AsyncExecutor asyncExecutor)
+ {
+ this.sidecarCdcOptions = sidecarCdcOptions;
+ this.cdcOptions = cdcOptions;
+ this.sidecarCdcStats = sidecarCdcStats;
+ this.cassandraClient = cassandraClient;
+ this.asyncExecutor = asyncExecutor;
+ }
+
+ // StatePersister implemented methods
+
+ @Override
+ public void persist(String jobId, int partitionId, @Nullable TokenRange tokenRange, @NotNull ByteBuffer buf)
+ {
+ final PersistWrapper latest = new PersistWrapper(jobId, partitionId, tokenRange, buf, timestampGenerator.next());
+ PersistWrapper.Key key = latest.key();
+ if (!latest.equals(this.latestState.get(key)))
+ {
+ this.latestState.compute(key, (k, prev) -> !latest.equals(prev) ? latest : prev);
+ }
+ }
+
+ @NotNull
+ @Override
+ public List<CdcState> loadState(String jobId, int partitionId, @Nullable TokenRange tokenRange)
+ {
+ CompressionUtil compressionUtil = CdcBridgeFactory.get(cdcOptions.version()).compressionUtil();
+ List<Integer> sizes = new ArrayList<>();
+ // deserialize and merge the CDC state objects into canonical view
+ List<CdcState> result = loadStateForRange(jobId, tokenRange)
+ .peek(bytes -> sizes.add(bytes.length))
+ .map(bytes -> CdcState.deserialize(CdcKryoRegister.kryo(), compressionUtil, bytes))
Review Comment:
nit:
```suggestion
.map(bytes -> {
sizes.add(bytes.length);
return CdcState.deserialize(CdcKryoRegister.kryo(), compressionUtil, bytes);
})
```
##########
cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java:
##########
@@ -122,51 +119,6 @@ public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvid
return buildClient(sidecarConfig, vertx, httpClientConfig, sidecarInstancesProvider);
}
- static String transportModeBasedWriterUserAgent(DataTransport transport)
- {
- switch (transport)
- {
- case S3_COMPAT:
- return BuildInfo.WRITER_S3_USER_AGENT;
- case DIRECT:
- default:
- return BuildInfo.WRITER_USER_AGENT;
- }
- }
-
- public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvider, BulkSparkConf conf)
- {
- Vertx vertx = Vertx.vertx(new VertxOptions().setUseDaemonThread(true)
- .setWorkerPoolSize(conf.getMaxHttpConnections()));
-
Review Comment:
Just curious, what prevent from having it kept in the same file? (instead of moving to `AnalyticsSidecarClient`). I think the idea of further consolidate.
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarStatePersister.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.ThreadLocalMonotonicTimestampGenerator;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
+import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.cdc.CdcKryoRegister;
+import org.apache.cassandra.cdc.api.CdcOptions;
+import org.apache.cassandra.cdc.api.StatePersister;
+import org.apache.cassandra.cdc.state.CdcState;
+import org.apache.cassandra.spark.utils.AsyncExecutor;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.util.CompressionUtil;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * SidecarStatePersister buffers CDC state and flushes at regular time intervals, so we only write the latest CDC state and don't wastefully write expired data.
+ */
+public class SidecarStatePersister implements StatePersister
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(SidecarStatePersister.class);
+
+ // group latest state by jobId/token range, so we persist independently
+ protected final ConcurrentHashMap<PersistWrapper.Key, PersistWrapper> latestState = new ConcurrentHashMap<>();
+ protected final ConcurrentLinkedQueue<TimedFutureWrapper> activeFlush = new ConcurrentLinkedQueue<>();
+ private final ThreadLocalMonotonicTimestampGenerator timestampGenerator = new ThreadLocalMonotonicTimestampGenerator();
+ private final SidecarCdcOptions sidecarCdcOptions;
+ private final CdcOptions cdcOptions;
+ private final SidecarCdcCassandraClient cassandraClient;
+ private final SidecarCdcStats sidecarCdcStats;
+ private final AsyncExecutor asyncExecutor;
+ volatile long timerId = -1L;
+
+ public SidecarStatePersister(SidecarCdcOptions sidecarCdcOptions,
+ CdcOptions cdcOptions,
+ SidecarCdcStats sidecarCdcStats,
+ SidecarCdcCassandraClient cassandraClient,
+ AsyncExecutor asyncExecutor)
+ {
+ this.sidecarCdcOptions = sidecarCdcOptions;
+ this.cdcOptions = cdcOptions;
+ this.sidecarCdcStats = sidecarCdcStats;
+ this.cassandraClient = cassandraClient;
+ this.asyncExecutor = asyncExecutor;
+ }
+
+ // StatePersister implemented methods
+
+ @Override
+ public void persist(String jobId, int partitionId, @Nullable TokenRange tokenRange, @NotNull ByteBuffer buf)
+ {
+ final PersistWrapper latest = new PersistWrapper(jobId, partitionId, tokenRange, buf, timestampGenerator.next());
+ PersistWrapper.Key key = latest.key();
+ if (!latest.equals(this.latestState.get(key)))
+ {
+ this.latestState.compute(key, (k, prev) -> !latest.equals(prev) ? latest : prev);
+ }
+ }
+
+ @NotNull
+ @Override
+ public List<CdcState> loadState(String jobId, int partitionId, @Nullable TokenRange tokenRange)
+ {
+ CompressionUtil compressionUtil = CdcBridgeFactory.get(cdcOptions.version()).compressionUtil();
+ List<Integer> sizes = new ArrayList<>();
+ // deserialize and merge the CDC state objects into canonical view
+ List<CdcState> result = loadStateForRange(jobId, tokenRange)
+ .peek(bytes -> sizes.add(bytes.length))
+ .map(bytes -> CdcState.deserialize(CdcKryoRegister.kryo(), compressionUtil, bytes))
+ .collect(Collectors.toList());
+ int count = sizes.size();
+ int len = sizes.stream().mapToInt(i -> i).sum();
+ LOGGER.debug("Read CDC state from Cassandra jobId={} start={} end={} stateCount={} stateSize={}",
+ jobId, tokenRange == null ? "null" : tokenRange.lowerEndpoint(), tokenRange == null ? "null" : tokenRange.upperEndpoint(), count, len);
+ sidecarCdcStats.captureCdcConsumerReadFromState(count, len);
+ return result;
+ }
+
+ @VisibleForTesting
+ public Stream<byte[]> loadStateForRange(String jobId, @Nullable TokenRange tokenRange)
+ {
+ return cassandraClient
+ .loadStateForRange(jobId, tokenRange);
+ }
+
+ /**
+ * Start the SidecarStatePersister to flush to Cassandra every `persistDelay()`
+ */
+ public synchronized void start()
+ {
+ if (timerId >= 0)
+ {
+ // already running
+ return;
+ }
+ this.timerId = asyncExecutor.periodicTimer(this::persistToCassandra, sidecarCdcOptions.persistDelay().toMillis());
+ }
+
+ /**
+ * Stop the SidecarStatePersister gracefully, blocking to await for any pending flushes to complete.
+ */
+ public void stop()
+ {
+ stop(true);
+ }
+
+ public synchronized void stop(boolean flush)
+ {
+ if (this.timerId < 0)
+ {
+ // not running
+ return;
+ }
+
+ asyncExecutor.cancelTimer(this.timerId);
+ this.timerId = -1;
Review Comment:
Relying on the condition `timerId == -1` might not be sufficient. The scheduled run could be still running when cancelTimer returns. However, it is not something to be addressed in this patch. Just calling it out.
##########
cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/CdcBuilder.java:
##########
@@ -41,26 +41,26 @@
public class CdcBuilder
{
@NotNull
- final String jobId;
- final int partitionId;
+ protected final String jobId;
+ protected final int partitionId;
@NotNull
TokenRangeSupplier tokenRangeSupplier = () -> null;
@NotNull
- SchemaSupplier schemaSupplier;
+ protected SchemaSupplier schemaSupplier;
@NotNull
CassandraSource cassandraSource = CassandraSource.DEFAULT;
@NotNull
StatePersister statePersister = StatePersister.STUB;
@NotNull
- CdcOptions cdcOptions = CdcOptions.DEFAULT;
+ protected CdcOptions cdcOptions = CdcOptions.DEFAULT;
@NotNull
ICdcStats stats = ICdcStats.STUB;
@Nullable
- AsyncExecutor asyncExecutor = null;
+ protected AsyncExecutor asyncExecutor = null;
@Nullable
- CommitLogProvider commitLogProvider = null;
+ protected CommitLogProvider commitLogProvider = null;
@NotNull
- EventConsumer eventConsumer;
+ protected EventConsumer eventConsumer;
@NotNull
Review Comment:
Those are `@NotNull`
##########
cassandra-analytics-cdc-sidecar/src/test/java/org/apache/cassandra/cdc/sidecar/SidecarCommitLogProviderTests.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.math.BigInteger;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.commons.lang3.mutable.MutableInt;
+import org.junit.jupiter.api.Test;
+
+import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.cdc.api.CommitLog;
+import org.apache.cassandra.spark.data.ReplicationFactor;
+import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
+import org.apache.cassandra.spark.data.partitioner.Partitioner;
+import org.jetbrains.annotations.NotNull;
+import org.mockito.stubbing.Answer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class SidecarCommitLogProviderTests
+{
+ private static final List<CassandraInstance> INSTANCES = Arrays.asList(
+ new CassandraInstance("0", "local1", "DC1"),
+ new CassandraInstance("100", "local2", "DC1"),
+ new CassandraInstance("200", "local3", "DC1"),
+ new CassandraInstance("300", "local4", "DC1"),
+ new CassandraInstance("400", "local5", "DC1"),
+ new CassandraInstance("500", "local6", "DC1"),
+ new CassandraInstance("1", "local7", "DC2"),
+ new CassandraInstance("101", "local8", "DC2"),
+ new CassandraInstance("201", "local9", "DC2"),
+ new CassandraInstance("301", "local10", "DC2"),
+ new CassandraInstance("401", "local11", "DC2"),
+ new CassandraInstance("501", "local12", "DC2")
+ );
+
+ @Test
+ public void testCommitLogProvider()
+ {
+ CassandraInstance instance1 = INSTANCES.get(0);
+ CassandraInstance instance2 = INSTANCES.get(1);
+ CassandraInstance instance3 = INSTANCES.get(2);
+ CassandraInstance instance4 = INSTANCES.get(3);
+ CassandraInstance instance5 = INSTANCES.get(4);
+ CassandraInstance instance6 = INSTANCES.get(5);
+
+ ClusterConfigProvider clusterConfigProvider = getClusterConfigProvider();
+ SidecarCdcClient sidecarCdcClient = mock(SidecarCdcClient.class);
+ Map<CassandraInstance, Integer> instanceListCount = new HashMap<>();
+ MutableInt segmentIdGenerator = new MutableInt(0);
+ Answer<CompletableFuture<List<CommitLog>>> listCommitLogAnswer = invocation -> {
+ CassandraInstance instance = invocation.getArgument(0, CassandraInstance.class);
+ instanceListCount.compute(instance, (key, prev) -> prev == null ? 1 : prev + 1);
+ int instanceId = Integer.parseInt(instance.nodeName().substring(5));
+ int numLogs = instanceId * 13;
+ List<CommitLog> logs = IntStream.rangeClosed(1, numLogs)
+ .mapToObj(segmentId -> mockCommitLog(instance, segmentIdGenerator.getAndAdd(1)))
+ .collect(Collectors.toList());
+ return CompletableFuture.completedFuture(logs);
+ };
+
+ when(sidecarCdcClient.listCdcCommitLogSegments(any(CassandraInstance.class))).thenAnswer(listCommitLogAnswer);
+ SidecarCommitLogProvider commitLogProvider = getSidecarCommitLogProvider(clusterConfigProvider, sidecarCdcClient);
+
+ // list logs on instance1, instance2, instance3
+ TokenRange tokenRange = TokenRange.openClosed(
+ new BigInteger("-50"),
+ new BigInteger("-10")
+ );
+ List<CommitLog> logs = commitLogProvider
+ .logs(tokenRange)
+ .sorted()
+ .collect(Collectors.toList());
+ int expectedNumberOfLogs = 13 + (2 * 13) + (3 * 13);
+ assertEquals(expectedNumberOfLogs, logs.size());
+ IntStream.range(0, logs.size())
+ .forEach(index -> assertEquals(index, logs.get(index).segmentId()));
+ assertEquals(3, instanceListCount.size());
+ assertEquals(1, instanceListCount.get(instance1));
+ assertEquals(1, instanceListCount.get(instance2));
+ assertEquals(1, instanceListCount.get(instance3));
+
+ // list logs on instance4, instance5, instance6
+ TokenRange tokenRange2 = TokenRange.openClosed(
+ new BigInteger("210"),
+ new BigInteger("280")
+ );
+ List<CommitLog> logs2 = commitLogProvider
+ .logs(tokenRange2)
+ .collect(Collectors.toList());
+ expectedNumberOfLogs = (4 * 13) + (5 * 13) + (6 * 13);
Review Comment:
nit: extract "4, 5, 6 and 13" to variable to improve readability.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r1963680521
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcCassandraClient.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Stream;
+
+import com.datastax.driver.core.ResultSetFuture;
+import org.apache.cassandra.bridge.TokenRange;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Interface defining read and write operations to Cassandra via the `SidecarStatePersister`
Review Comment:
`SidecarCdcCassandraClient ` is used by `SidecarStatePersister`
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2003620243
##########
scripts/build-sidecar.sh:
##########
@@ -44,7 +44,7 @@ if [ "x${LOCAL_SIDECAR_REPO}" != "x" ]; then
else
SIDECAR_REPO="${SIDECAR_REPO:-https://github.com/apache/cassandra-sidecar.git}"
SIDECAR_BRANCH="${SIDECAR_BRANCH:-trunk}"
- SIDECAR_COMMIT="${SIDECAR_COMMIT:-f3bcbba3dcd81b640711baa35a76a2d949ce6c5e}"
+ SIDECAR_COMMIT="${SIDECAR_COMMIT:-c1260886642e789fa15a3fb900251bea529fcc09}"
Review Comment:
This is fixed after resolving the conflicts with `trunk`.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2001356175
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarStatePersister.java:
##########
@@ -0,0 +1,434 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.ThreadLocalMonotonicTimestampGenerator;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
+import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.cdc.CdcKryoRegister;
+import org.apache.cassandra.cdc.api.CdcOptions;
+import org.apache.cassandra.cdc.api.StatePersister;
+import org.apache.cassandra.cdc.state.CdcState;
+import org.apache.cassandra.spark.utils.AsyncExecutor;
+import org.apache.cassandra.spark.utils.ThrowableUtils;
+import org.apache.cassandra.util.CompressionUtil;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * SidecarStatePersister buffers CDC state and flushes at regular time intervals, so we only write the latest CDC state and don't wastefully write expired data.
+ */
+public class SidecarStatePersister implements StatePersister
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(SidecarStatePersister.class);
+
+ // group latest state by jobId/token range, so we persist independently
+ protected final ConcurrentHashMap<PersistWrapper.Key, PersistWrapper> latestState = new ConcurrentHashMap<>();
+ protected final ConcurrentLinkedQueue<TimedFutureWrapper> activeFlush = new ConcurrentLinkedQueue<>();
+ private final ThreadLocalMonotonicTimestampGenerator timestampGenerator = new ThreadLocalMonotonicTimestampGenerator();
+ private final SidecarCdcOptions sidecarCdcOptions;
+ private final CdcOptions cdcOptions;
+ private final SidecarCdcCassandraClient cassandraClient;
+ private final SidecarCdcStats sidecarCdcStats;
+ private final AsyncExecutor asyncExecutor;
+ volatile long timerId = -1L;
+
+ public SidecarStatePersister(SidecarCdcOptions sidecarCdcOptions,
+ CdcOptions cdcOptions,
+ SidecarCdcStats sidecarCdcStats,
+ SidecarCdcCassandraClient cassandraClient,
+ AsyncExecutor asyncExecutor)
+ {
+ this.sidecarCdcOptions = sidecarCdcOptions;
+ this.cdcOptions = cdcOptions;
+ this.sidecarCdcStats = sidecarCdcStats;
+ this.cassandraClient = cassandraClient;
+ this.asyncExecutor = asyncExecutor;
+ }
+
+ // StatePersister implemented methods
+
+ @Override
+ public void persist(String jobId, int partitionId, @Nullable TokenRange tokenRange, @NotNull ByteBuffer buf)
+ {
+ final PersistWrapper latest = new PersistWrapper(jobId, partitionId, tokenRange, buf, timestampGenerator.next());
+ PersistWrapper.Key key = latest.key();
+ if (!latest.equals(this.latestState.get(key)))
+ {
+ this.latestState.compute(key, (k, prev) -> !latest.equals(prev) ? latest : prev);
+ }
+ }
+
+ @NotNull
+ @Override
+ public List<CdcState> loadState(String jobId, int partitionId, @Nullable TokenRange tokenRange)
+ {
+ CompressionUtil compressionUtil = CdcBridgeFactory.get(cdcOptions.version()).compressionUtil();
+ List<Integer> sizes = new ArrayList<>();
+ // deserialize and merge the CDC state objects into canonical view
+ List<CdcState> result = loadStateForRange(jobId, tokenRange)
+ .peek(bytes -> sizes.add(bytes.length))
+ .map(bytes -> CdcState.deserialize(CdcKryoRegister.kryo(), compressionUtil, bytes))
Review Comment:
I mean...how does that improve it?
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r2001374156
##########
cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/CdcBuilder.java:
##########
@@ -41,26 +41,26 @@
public class CdcBuilder
{
@NotNull
- final String jobId;
- final int partitionId;
+ protected final String jobId;
+ protected final int partitionId;
@NotNull
TokenRangeSupplier tokenRangeSupplier = () -> null;
@NotNull
- SchemaSupplier schemaSupplier;
+ protected SchemaSupplier schemaSupplier;
@NotNull
CassandraSource cassandraSource = CassandraSource.DEFAULT;
@NotNull
StatePersister statePersister = StatePersister.STUB;
@NotNull
- CdcOptions cdcOptions = CdcOptions.DEFAULT;
+ protected CdcOptions cdcOptions = CdcOptions.DEFAULT;
@NotNull
ICdcStats stats = ICdcStats.STUB;
@Nullable
- AsyncExecutor asyncExecutor = null;
+ protected AsyncExecutor asyncExecutor = null;
@Nullable
- CommitLogProvider commitLogProvider = null;
+ protected CommitLogProvider commitLogProvider = null;
@NotNull
- EventConsumer eventConsumer;
+ protected EventConsumer eventConsumer;
@NotNull
Review Comment:
Yes.
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "jberragan (via GitHub)" <gi...@apache.org>.
jberragan commented on PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#issuecomment-2744159343
Latest green CircleCI: https://app.circleci.com/pipelines/github/jberragan/cassandra-analytics/187/workflows/67edad7c-1294-413c-8db7-dfc49bccd6ca
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
Re: [PR] CASSANALYTICS-7: Add Apache Cassandra Sidecar CDC implementation [cassandra-analytics]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #99:
URL: https://github.com/apache/cassandra-analytics/pull/99#discussion_r1962540747
##########
cassandra-analytics-cdc-sidecar/build.gradle:
##########
@@ -0,0 +1,142 @@
+import java.nio.file.Paths
+
+/*
+ * 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.
+ */
+
+plugins {
+ id('java-library')
+ id('maven-publish')
+}
+
+configurations {
+ all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
+ all*.exclude(group: 'log4j', module: 'log4j')
+}
+
+java {
+ withJavadocJar()
+ withSourcesJar()
+}
+
+publishing {
+ publications {
+ maven(MavenPublication) {
+ from components.java
+ groupId project.group
+ artifactId "${archivesBaseName}"
+ version System.getenv("CODE_VERSION") ?: "${version}"
+ }
+ }
+}
+
+configurations {
+ fourzero {
+ description = 'Cassandra 4.0 dependency'
+ }
+ fourzerobridge {
+ description = 'Bridge to Cassandra 4.0'
+ }
+ fourzerotypes {
+ description = 'Cassandra 4.0 Data Types'
+ }
+}
+
+dependencies {
+ api(project(':cassandra-bridge'))
+
+ fourzero(project(path: ':cassandra-four-zero', configuration: 'shadow'))
+ fourzerobridge(project(path: ':cassandra-four-zero-bridge'))
+ fourzerotypes(project(path: ':cassandra-four-zero-types'))
+
+ implementation project(":cassandra-analytics-common")
+ implementation project(':cassandra-analytics-cdc')
+ implementation project(':cassandra-analytics-sidecar-client')
+ compileOnlyApi("com.datastax.cassandra:cassandra-driver-core:${project.cassandraDriverVersion}") // Cassandra driver is required only if user supplies the CassandraClientSource
+
+ implementation "com.google.guava:guava:${project.guavaVersion}"
+
+ implementation "org.apache.commons:commons-lang3:${project.commonsLang3Version}"
+ implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
+ implementation "com.esotericsoftware:kryo-shaded:${kryoVersion}"
+ implementation "com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}"
+ implementation "com.fasterxml.jackson.core:jackson-core:${jacksonVersion}"
+ implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
+
+ testImplementation project(":cassandra-analytics-common")
+
+ // pull in cassandra-bridge so we can re-use TestSchema to generate arbitrary schemas for the cdc tests
+ testImplementation project(":cassandra-bridge")
+ testImplementation("com.datastax.cassandra:cassandra-driver-core:${project.cassandraDriverVersion}")
+ testImplementation(testFixtures(project(':cassandra-bridge')))
+ testImplementation project(":cassandra-four-zero-bridge")
+ testImplementation project(":cassandra-four-zero-types")
+ testImplementation project(path: ':cassandra-four-zero', configuration: 'shadow')
+ testImplementation(group: 'org.quicktheories', name: 'quicktheories', version: "${project.rootProject.quickTheoriesVersion}")
+ testImplementation(group: 'com.google.guava', name: 'guava', version: '31.1-jre')
Review Comment:
Can it use `${guavaVersion}`?
##########
cassandra-analytics-cdc-sidecar/build.gradle:
##########
@@ -0,0 +1,142 @@
+import java.nio.file.Paths
+
+/*
+ * 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.
+ */
+
+plugins {
+ id('java-library')
+ id('maven-publish')
+}
+
+configurations {
+ all*.exclude(group: 'org.slf4j', module: 'slf4j-log4j12')
+ all*.exclude(group: 'log4j', module: 'log4j')
+}
+
+java {
+ withJavadocJar()
+ withSourcesJar()
+}
+
+publishing {
+ publications {
+ maven(MavenPublication) {
+ from components.java
+ groupId project.group
+ artifactId "${archivesBaseName}"
+ version System.getenv("CODE_VERSION") ?: "${version}"
+ }
+ }
+}
+
+configurations {
+ fourzero {
+ description = 'Cassandra 4.0 dependency'
+ }
+ fourzerobridge {
+ description = 'Bridge to Cassandra 4.0'
+ }
+ fourzerotypes {
+ description = 'Cassandra 4.0 Data Types'
+ }
+}
+
+dependencies {
+ api(project(':cassandra-bridge'))
+
+ fourzero(project(path: ':cassandra-four-zero', configuration: 'shadow'))
+ fourzerobridge(project(path: ':cassandra-four-zero-bridge'))
+ fourzerotypes(project(path: ':cassandra-four-zero-types'))
+
+ implementation project(":cassandra-analytics-common")
+ implementation project(':cassandra-analytics-cdc')
+ implementation project(':cassandra-analytics-sidecar-client')
+ compileOnlyApi("com.datastax.cassandra:cassandra-driver-core:${project.cassandraDriverVersion}") // Cassandra driver is required only if user supplies the CassandraClientSource
+
+ implementation "com.google.guava:guava:${project.guavaVersion}"
+
+ implementation "org.apache.commons:commons-lang3:${project.commonsLang3Version}"
+ implementation "org.slf4j:slf4j-api:${slf4jApiVersion}"
+ implementation "com.esotericsoftware:kryo-shaded:${kryoVersion}"
+ implementation "com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}"
+ implementation "com.fasterxml.jackson.core:jackson-core:${jacksonVersion}"
+ implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
+
+ testImplementation project(":cassandra-analytics-common")
+
+ // pull in cassandra-bridge so we can re-use TestSchema to generate arbitrary schemas for the cdc tests
+ testImplementation project(":cassandra-bridge")
+ testImplementation("com.datastax.cassandra:cassandra-driver-core:${project.cassandraDriverVersion}")
+ testImplementation(testFixtures(project(':cassandra-bridge')))
+ testImplementation project(":cassandra-four-zero-bridge")
+ testImplementation project(":cassandra-four-zero-types")
+ testImplementation project(path: ':cassandra-four-zero', configuration: 'shadow')
+ testImplementation(group: 'org.quicktheories', name: 'quicktheories', version: "${project.rootProject.quickTheoriesVersion}")
+ testImplementation(group: 'com.google.guava', name: 'guava', version: '31.1-jre')
+ testImplementation(group: 'com.github.luben', name: 'zstd-jni', version: '1.5.0-4')
+ testImplementation(group: 'org.slf4j', name: 'slf4j-simple', version: '1.7.26')
+ testImplementation "org.quicktheories:quicktheories:${project.rootProject.quickTheoriesVersion}"
+ testImplementation "org.junit.jupiter:junit-jupiter-api:${project.junitVersion}"
+ testImplementation "org.junit.jupiter:junit-jupiter-params:${project.junitVersion}"
+ testImplementation "org.junit.jupiter:junit-jupiter-engine:${project.junitVersion}"
+ testImplementation "org.mockito:mockito-core:${project.rootProject.mockitoVersion}"
+ testImplementation "org.mockito:mockito-inline:${project.rootProject.mockitoVersion}"
+}
+
+jar {
+ dependsOn(configurations.fourzero)
+ dependsOn(configurations.fourzerobridge)
+ dependsOn(configurations.fourzerotypes)
+
+// writeBuildVersion(version, project.projectDir)
+
+ exclude('META-INF/*')
+ into('bridges') {
+ from(configurations.fourzero.singleFile)
+ from(configurations.fourzerobridge.singleFile)
+ from(configurations.fourzerotypes.singleFile)
+ }
+}
+
+//private void writeBuildVersion(version, projectDir) {
Review Comment:
Not needed and `cassandra-analytics-cdc-build.properties` too.
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/ClusterConfigProvider.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.util.Set;
+
+import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
+import org.apache.cassandra.spark.data.partitioner.Partitioner;
+
+public interface ClusterConfigProvider
+{
+ String dc();
+
+ Set<CassandraInstance> getCluster();
+
+ Partitioner partitioner();
+}
Review Comment:
Add javadocs
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcCassandraClient.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Stream;
+
+import com.datastax.driver.core.ResultSetFuture;
+import org.apache.cassandra.bridge.TokenRange;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Interface defining read and write operations to Cassandra via the `SidecarStatePersister`
Review Comment:
`SidecarStatePersister` is not reference anywhere else in this file.
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarDownMonitor.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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.cassandra.cdc.sidecar;
+
+import org.apache.cassandra.spark.data.partitioner.CassandraInstance;
+
+public interface SidecarDownMonitor
Review Comment:
javadoc
--
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: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org