Javadoc, Dependencyies, test

Author: clun

pom.xml

@@ -13,14 +13,15 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <maven.plugin.compiler.source>11</maven.plugin.compiler.source>
     <maven.plugin.compiler.target>11</maven.plugin.compiler.target>
+    <maven.skip.test>false</maven.skip.test>
     <!-- Beam -->
     <beam.version>2.48.0</beam.version>
     <!-- Astra -->
-    <astra-sdk.version>0.6</astra-sdk.version>
+    <astra-sdk.version>0.6.2</astra-sdk.version>
     <cassandra-driver4x.version>4.16.0</cassandra-driver4x.version>
     <!-- Third Parties -->
     <slf4j.version>2.0.7</slf4j.version>
-    <logback.version>1.4.7</logback.version>
+    <logback.version>1.4.8</logback.version>
     <auto-value.version>1.10.1</auto-value.version>
     <junit.version>4.13.2</junit.version>
     <hamcrest.version>2.2</hamcrest.version>
@@ -31,9 +32,9 @@
     <version.maven.plugin.exec>1.6.0</version.maven.plugin.exec>
     <version.maven.plugin.gpg>3.0.1</version.maven.plugin.gpg>
     <version.maven.plugin.jar>3.3.0</version.maven.plugin.jar>
-    <version.maven.plugin.javadoc>3.4.1</version.maven.plugin.javadoc>
+    <version.maven.plugin.javadoc>3.5.0</version.maven.plugin.javadoc>
     <version.maven.plugin.license>2.0.0</version.maven.plugin.license>
-    <version.maven.plugin.nexus>1.6.8</version.maven.plugin.nexus>
+    <version.maven.plugin.nexus>1.6.13</version.maven.plugin.nexus>
     <version.maven.plugin.release>2.5.2</version.maven.plugin.release>
     <version.maven.plugin.resources>3.3.1</version.maven.plugin.resources>
     <version.maven.plugin.shade>3.4.1</version.maven.plugin.shade>
@@ -185,7 +186,6 @@
         </configuration>
       </plugin>
 
-      <!--
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-javadoc-plugin</artifactId>
@@ -199,18 +199,16 @@
           </execution>
         </executions>
       </plugin>
-      -->
 
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-surefire-plugin</artifactId>
         <version>${version.maven.plugin.surefire}</version>
         <configuration>
-          <skipTests>true</skipTests>
+          <skipTests>${maven.skip.test}</skipTests>
         </configuration>
       </plugin>
 
-
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-dependency-plugin</artifactId>

src/main/java/org/apache/beam/sdk/io/astra/db/AstraDbIO.java

@@ -141,6 +141,12 @@ public abstract static class Read<T> extends PTransform<PBegin, PCollection<T>>
 
     abstract @Nullable Coder<T> coder();
 
+    /**
+     * Expose split count.
+     *
+     * @return
+     *    minimal split count
+     */
     public abstract @Nullable ValueProvider<Integer> minNumberOfSplits();
 
     abstract @Nullable SerializableFunction<CqlSession, AstraDbMapper<T>> mapperFactoryFn();
@@ -853,6 +859,9 @@ public PCollection<T> expand(PCollection<Read<T>> input) {
     }
   }
 
+  /**
+   * Closing Open Connections.
+   */
   public static void close() {
     CqlSessionHolder.cleanup();
   }

src/main/java/org/apache/beam/sdk/io/astra/db/CqlSessionHolder.java

@@ -96,6 +96,8 @@ public static synchronized CqlSession getCqlSession(AstraDbIO.Write<?> write) {
    *    token (or clientSecret)
    * @param secureConnectBundle
    *    read scb as stream
+   * @param keyspace
+   *    list of keyspaces
    * @return
    *    cassandra cluster
    */
@@ -172,6 +174,9 @@ private static synchronized void init() {
     }
   }
 
+  /**
+   * Close all sessions.
+   */
   public static void cleanup() {
     cacheSessions.values().stream()
             .filter(s->!s.isClosed())

src/main/java/org/apache/beam/sdk/io/astra/db/ExecuteCqlFn.java

@@ -29,27 +29,60 @@
  */
 public abstract class ExecuteCqlFn<T> extends DoFn<String, Row> {
 
+    /**
+     * The Cassandra session.
+     */
     CqlSession cqlSession;
 
+    /**
+     * Constructor with Session.
+     *
+     * @param session the Cassandra session.
+     */
     public ExecuteCqlFn(CqlSession session) {
         this.cqlSession = session;
     }
 
+    /**
+     * Constructor with read.
+     *
+     * @param read reader.
+     */
     public ExecuteCqlFn(AstraDbIO.Read<?> read) {
         this.cqlSession = CqlSessionHolder.getCqlSession(
                 read.token(), read.secureConnectBundle(), read.keyspace());
     }
 
+    /**
+     * Constructor with write.
+     *
+     * @param write writer
+     */
     public ExecuteCqlFn(AstraDbIO.Write<?> write) {
         this.cqlSession = CqlSessionHolder.getCqlSession(
                 write.token(), write.secureConnectBundle(), write.keyspace());
     }
 
+    /**
+     * Executing query as implementation of the DoFn.
+     * @param query
+     *      current query
+     * @param receiver
+     *      query responses
+     */
     @ProcessElement
     public void processElement(@Element String query, OutputReceiver<T> receiver) {
         cqlSession.execute(query).forEach(row -> receiver.output(mapRow(row)));
     }
 
+    /**
+     * Show be implemented to map row.
+     *
+     * @param row
+     *      target row
+     * @return
+     *      row mapper
+     */
     public abstract T mapRow(Row row);
 
 

src/main/java/org/apache/beam/sdk/io/astra/db/ExecuteCqlSimpleFn.java

@@ -28,18 +28,41 @@
  */
 public class ExecuteCqlSimpleFn extends ExecuteCqlFn<Row> {
 
+    /**
+     * Constructor with Session.
+     *
+     * @param session the Cassandra session.
+     */
     public ExecuteCqlSimpleFn(CqlSession session) {
         super(session);
     }
 
+    /**
+     * Constructor with read.
+     *
+     * @param read the Cassandra session.
+     */
     public ExecuteCqlSimpleFn(AstraDbIO.Read<?> read) {
        super(read);
     }
 
+    /**
+     * Constructor with Write.
+     *
+     * @param write the Cassandra session.
+     */
     public ExecuteCqlSimpleFn(AstraDbIO.Write<?> write) {
         super(write);
     }
 
+    /**
+     * No Mapping.
+     *
+     * @param row
+     *      cassandra row
+     * @return
+     *      row
+     */
     @Override
     public Row mapRow(Row row) {
         return row;

src/main/java/org/apache/beam/sdk/io/astra/db/mapping/AstraDbMapper.java

@@ -51,6 +51,13 @@
  */
 public interface AstraDbMapper<T> {
 
+  /**
+   * Produce object out of the row
+   * @param row
+   *    cassandra row
+   * @return
+   *    entity
+   */
   @GetEntity
   T mapRow(Row row);
 

src/main/java/org/apache/beam/sdk/io/astra/db/mapping/BeamRowDbMapper.java

@@ -86,9 +86,15 @@ public class BeamRowDbMapper implements AstraDbMapper<Row>, Serializable {
      */
     private final Set<String> primaryKeysColumnNames;
 
-    List<ColumnDefinition> columnDefinitions;
+    /**
+     * List of columns of the current table.
+     */
+    private List<ColumnDefinition> columnDefinitions;
 
-    Schema beamSchema;
+    /**
+     * Associated bean schema
+     */
+    private Schema beamSchema;
 
     /**
      * Constructor used by AstraDbRowMapperFactory.

src/main/java/org/apache/beam/sdk/io/astra/db/transforms/split/AstraSplitFn.java

@@ -53,6 +53,13 @@ public AstraSplitFn() {
         LOG.info("Split into Token Ranges.");
     }
 
+    /**
+     * Implement splits in token ranges.
+     * @param read
+     *      current read
+     * @param outputReceiver
+     *      splits
+     */
     @ProcessElement
     public void process(@Element AstraDbIO.Read<T> read, OutputReceiver<AstraDbIO.Read<T>> outputReceiver) {
         getRingRanges(read)

src/main/java/org/apache/beam/sdk/io/astra/db/transforms/split/AstraTokenFactory.java

@@ -44,26 +44,60 @@
 import java.math.BigInteger;
 import java.util.Set;
 
+/**
+ * Produce Token Ranges on Mumur3 (used in AStra)
+ */
 public class AstraTokenFactory extends Murmur3TokenFactory {
 
+  /** Total number of tokens in the ring. */
   public static final BigInteger TOTAL_TOKEN_COUNT =
           BigInteger.valueOf(Long.MAX_VALUE).subtract(BigInteger.valueOf(Long.MIN_VALUE));
 
+  /**
+   * Get total number of token for the ring.
+   *
+   * @return
+   *    token number
+   */
   @NonNull
   public BigInteger totalTokenCount() {
     return TOTAL_TOKEN_COUNT;
   }
 
+  /**
+   * Compute a range of token.
+   *
+   * @param start
+   *    start token
+   * @param end
+   *    end token
+   * @param replicas
+   *    replicas endpoints
+   * @return
+   *    a token range
+   */
   @NonNull
   public AstraTokenRange range(@NonNull Token start, @NonNull Token end, @NonNull Set<AstraTokenRangeEndpoint> replicas) {
     return new AstraTokenRange(((Murmur3Token) start), (Murmur3Token) end, replicas);
   }
 
+  /**
+   * Create Splitter.
+   *
+   * @return
+   *    token range splitter
+   */
   @NonNull
   public TokenRangeSplitter splitter() {
     return new AstraTokenRangeSplitter();
   }
 
+  /**
+   * Create token clusterer.
+   *
+   * @return
+   *    clusterer.
+   */
   @NonNull
   public TokenRangeClusterer clusterer() {
     return new TokenRangeClusterer(this);

src/main/java/org/apache/beam/sdk/io/astra/db/transforms/split/AstraTokenRange.java

@@ -50,33 +50,74 @@
  */
 public class AstraTokenRange extends Murmur3TokenRange implements Serializable {
 
+  /** Set of replicas. */
   private final Set<AstraTokenRangeEndpoint> replicas;
 
+  /**
+   * Constructor without replicas.
+   * @param start
+   *      start token
+   * @param end
+   *      end token
+   */
   public AstraTokenRange(@NonNull Murmur3Token start, @NonNull Murmur3Token end) {
     super(start, end);
     this.replicas = ImmutableSet.of();
   }
 
+  /**
+   * Full constructor.
+   * @param start
+   *    start token
+   * @param end
+   *    end token
+   * @param replicas
+   *    replicas
+   */
   public AstraTokenRange(@NonNull Murmur3Token start, @NonNull Murmur3Token end, @NonNull Set<AstraTokenRangeEndpoint> replicas) {
     super(start, end);
     this.replicas = ImmutableSet.copyOf(replicas);
   }
 
+  /**
+   * Start token
+   *
+   * @return
+   *   getter for start token
+   */
   @NonNull
   public Murmur3Token getStart() {
     return (Murmur3Token) super.getStart();
   }
 
+  /**
+   * End token.
+   *
+   * @return
+   *   getter for end token
+   */
   @NonNull
   public Murmur3Token getEnd() {
     return (Murmur3Token) super.getEnd();
   }
 
+  /**
+   * List of replicas.
+   *
+   * @return
+   *    replicas list
+   */
   @NonNull
   public Set<AstraTokenRangeEndpoint> replicas() {
     return replicas;
   }
 
+  /**
+   * Size of the range.
+   *
+   * @return
+   *    size
+   */
   @NonNull
   public BigInteger size() {
     BigInteger left = BigInteger.valueOf(getStart().getValue());
@@ -88,6 +129,12 @@ public BigInteger size() {
     }
   }
 
+  /**
+   * Fraction of ring range.
+   *
+   * @return
+   *    fracction of the range
+   */
   public double fraction() {
     return size().doubleValue() / AstraTokenFactory.TOTAL_TOKEN_COUNT.doubleValue();
   }