[issue-60] Pravega txn API changes

Author: EronWright

.gitattributes

@@ -7,6 +7,7 @@
 *.java text
 *.xml text
 *.sh text
+*.properties text
 
 # Files that should keep CRLF line endings
 *.bat text eol=crlf

gradle.properties

@@ -14,7 +14,7 @@ mockitoVersion=1.10.19
 
 # Version and base tags can be overridden at build time.
 connectorVersion=0.2.0-SNAPSHOT
-pravegaVersion=0.1.0
+pravegaVersion=0.2.0-SNAPSHOT
 
 # These properties are only needed for publishing to maven central
 # Pravega Signing Key

src/main/java/io/pravega/connectors/flink/FlinkExactlyOncePravegaWriter.java

@@ -70,7 +70,6 @@ public class FlinkExactlyOncePravegaWriter<T>
     private final String streamName;
 
     private final long txnTimeoutMillis;
-    private final long txnMaxTimeMillis;
     private final long txnGracePeriodMillis;
 
     // ----------- runtime fields -----------
@@ -102,7 +101,7 @@ public FlinkExactlyOncePravegaWriter(
             final PravegaEventRouter<T> router) {
 
         this(controllerURI, scope, streamName, serializationSchema, router,
-                DEFAULT_TXN_TIMEOUT_MILLIS, DEFAULT_TXN_TIMEOUT_MILLIS, DEFAULT_TX_SCALE_GRACE_MILLIS);
+                DEFAULT_TXN_TIMEOUT_MILLIS, DEFAULT_TX_SCALE_GRACE_MILLIS);
     }
 
 
@@ -115,8 +114,6 @@ public FlinkExactlyOncePravegaWriter(
      * @param serializationSchema   The implementation for serializing every event into pravega's storage format.
      * @param router                The implementation to extract the partition key from the event.
      * @param txnTimeoutMillis      The number of milliseconds after which the transaction will be aborted.
-     * @param txnMaxTimeMillis      The maximum time (in milliseconds) to which transaction timeout may be
-     *                              increased via the pingTransaction API.
      * @param txnGracePeriodMillis  The maximum amount of time, in milliseconds, until which transaction may
      *                              remain active, after a scale operation has been initiated on the underlying stream.
      */
@@ -127,7 +124,6 @@ public FlinkExactlyOncePravegaWriter(
             final SerializationSchema<T> serializationSchema,
             final PravegaEventRouter<T> router,
             final long txnTimeoutMillis,
-            final long txnMaxTimeMillis,
             final long txnGracePeriodMillis) {
 
         Preconditions.checkNotNull(controllerURI, "controllerURI");
@@ -136,7 +132,6 @@ public FlinkExactlyOncePravegaWriter(
         Preconditions.checkNotNull(serializationSchema, "serializationSchema");
         Preconditions.checkNotNull(router, "router");
         Preconditions.checkArgument(txnTimeoutMillis > 0, "txnTimeoutMillis must be > 0");
-        Preconditions.checkArgument(txnMaxTimeMillis > 0, "txnMaxTimeMillis must be > 0");
         Preconditions.checkArgument(txnGracePeriodMillis > 0, "txnGracePeriodMillis must be > 0");
 
         this.controllerURI = controllerURI;
@@ -146,7 +141,6 @@ public FlinkExactlyOncePravegaWriter(
         this.eventRouter = router;
 
         this.txnTimeoutMillis = txnTimeoutMillis;
-        this.txnMaxTimeMillis = txnMaxTimeMillis;
         this.txnGracePeriodMillis = txnGracePeriodMillis;
     }
 
@@ -160,13 +154,13 @@ public void open(Configuration parameters) throws Exception {
         this.pravegaWriter = clientFactory.createEventWriter(
                 this.streamName,
                 serializer,
-                EventWriterConfig.builder().build());
+                EventWriterConfig.builder().transactionTimeoutTime(txnTimeoutMillis).transactionTimeoutScaleGracePeriod(txnGracePeriodMillis).build());
 
         log.info("Initialized pravega writer for stream: {}/{} with controller URI: {}", this.scopeName,
                 this.streamName, this.controllerURI);
 
         // start the transaction that will hold the elements till the first checkpoint
-        this.currentTxn = this.pravegaWriter.beginTxn(txnTimeoutMillis, txnMaxTimeMillis, txnGracePeriodMillis);
+        this.currentTxn = this.pravegaWriter.beginTxn();
 
         log.debug("{} - started first transaction '{}'", name(), this.currentTxn.getTxnId());
 
@@ -227,7 +221,7 @@ public List<UUID> snapshotState(long checkpointId, long checkpointTime) throws E
         this.txnsPendingCommit.addLast(new TransactionAndCheckpoint<>(txn, checkpointId));
 
         // start the next transaction for what comes after this checkpoint
-        this.currentTxn = this.pravegaWriter.beginTxn(txnTimeoutMillis, txnMaxTimeMillis, txnGracePeriodMillis);
+        this.currentTxn = this.pravegaWriter.beginTxn();
 
         log.debug("{} - started new transaction '{}'", name(), this.currentTxn.getTxnId());
         log.debug("{} - storing pending transactions {}", name(), txnsPendingCommit);

src/test/java/io/pravega/connectors/flink/FlinkExactlyOncePravegaWriterTest.java

@@ -120,7 +120,6 @@ private static void runTest(
                         new IntSerializer(),
                         new IdentityRouter<>(),
                         30 * 1000,  // 30 secs timeout
-                        30 * 1000,
                         30 * 1000)
                 );