Merge pull request #963: Kafka watermarking

Author: je-ik

beam/core/src/test/java/cz/o2/proxima/beam/core/direct/io/BatchLogReadTest.java

@@ -42,6 +42,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.beam.runners.direct.DirectRunner;
@@ -66,6 +67,7 @@
 import org.apache.beam.sdk.values.PBegin;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.TypeDescriptors;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -139,6 +141,7 @@ public void testReadWithThroughputLimitDisabled() {
   }
 
   @Test(timeout = 60000)
+  @Ignore("Unreproducibly flaky with unknown purpose")
   public void testReadWithThroughputLimitWait() {
     int numElements = 1000;
     List<StreamElement> input = createInput(numElements);
@@ -323,14 +326,14 @@ private static ThroughputLimiter getThroughputLimiter() {
 
   private static ThroughputLimiter getThroughputLimiter(int waitInvocations) {
     return new ThroughputLimiter() {
-      long numInvocations = 0L;
+      final AtomicLong numInvocations = new AtomicLong();
 
       @Override
       public Duration getPauseTime(Context context) {
-        if (++numInvocations > waitInvocations) {
+        if (numInvocations.incrementAndGet() > waitInvocations) {
           return Duration.ofSeconds(5);
         }
-        // on first two invocations return zero
+        // on first N invocations return zero
         return Duration.ZERO;
       }
 

core/src/main/java/cz/o2/proxima/core/transaction/KeyAttributes.java

@@ -76,19 +76,19 @@ private KeyAttributes() {}
    * @param entity the entity descriptor
    * @param key the entity key
    * @param attributeDescriptor descriptor of wildcard or regular attribute
-   * @param sequenceId sequence ID of the read attribute
+   * @param sequentialId sequence ID of the read attribute
    */
   public static KeyAttribute ofAttributeDescriptor(
       EntityDescriptor entity,
       String key,
       AttributeDescriptor<?> attributeDescriptor,
-      long sequenceId) {
+      long sequentialId) {
 
     Preconditions.checkArgument(
         !attributeDescriptor.isWildcard(),
         "Please specify attribute suffix for wildcard attributes. Got attribute %s",
         attributeDescriptor);
-    return new KeyAttribute(entity, key, attributeDescriptor, sequenceId, false, null);
+    return new KeyAttribute(entity, key, attributeDescriptor, sequentialId, false, null);
   }
 
   /**
@@ -98,22 +98,22 @@ public static KeyAttribute ofAttributeDescriptor(
    * @param entity the entity descriptor
    * @param key the entity key
    * @param attributeDescriptor descriptor of wildcard or regular attribute
-   * @param sequenceId sequence ID of the read attribute
+   * @param sequentialId sequence ID of the read attribute
    * @param attributeSuffix a specific attribute suffix when {@code attributeDescriptor} is wildcard
    *     attribute
    */
   public static KeyAttribute ofAttributeDescriptor(
       EntityDescriptor entity,
       String key,
       AttributeDescriptor<?> attributeDescriptor,
-      long sequenceId,
+      long sequentialId,
       @Nullable String attributeSuffix) {
 
     Preconditions.checkArgument(
         !attributeDescriptor.isWildcard() || attributeSuffix != null,
         "Please specify attribute suffix for wildcard attributes. Got attribute %s",
         attributeDescriptor);
-    return new KeyAttribute(entity, key, attributeDescriptor, sequenceId, false, attributeSuffix);
+    return new KeyAttribute(entity, key, attributeDescriptor, sequentialId, false, attributeSuffix);
   }
 
   /**
@@ -166,7 +166,7 @@ public static KeyAttribute ofMissingAttribute(
   public static KeyAttribute ofStreamElement(StreamElement element) {
     Preconditions.checkArgument(
         element.hasSequentialId(),
-        "Elements read with enabled transactions need to use sequenceIds got %s.",
+        "Elements read with enabled transactions need to use sequentialIds got %s.",
         element);
     Preconditions.checkArgument(!element.isDeleteWildcard(), "Wildcard deletes not yet supported");
     return new KeyAttribute(

direct/core/src/main/java/cz/o2/proxima/direct/core/randomaccess/KeyValue.java

@@ -157,7 +157,7 @@ private KeyValue(
   private KeyValue(
       EntityDescriptor entityDesc,
       AttributeDescriptor<T> attrDesc,
-      long sequenceId,
+      long sequentialId,
       String key,
       String attribute,
       RandomOffset offset,
@@ -168,7 +168,7 @@ private KeyValue(
     super(
         entityDesc,
         attrDesc,
-        sequenceId,
+        sequentialId,
         key,
         attribute,
         stamp,

direct/core/src/main/java/cz/o2/proxima/direct/core/transaction/TransactionalOnlineAttributeWriter.java

@@ -172,7 +172,7 @@ public class Transaction implements AutoCloseable {
     @Getter private final String transactionId;
     private boolean commitAttempted = false;
     @Getter private State.Flags state;
-    private long sequenceId = -1L;
+    private long sequentialId = -1L;
     private long stamp = Long.MIN_VALUE;
     private final List<CompletableFuture<?>> runningUpdates =
         Collections.synchronizedList(new ArrayList<>());
@@ -289,7 +289,7 @@ private CompletableFuture<Response> sendCommitRequest(Collection<StreamElement>
         if (state != State.Flags.COMMITTED) {
           return CompletableFuture.completedFuture(Response.empty().aborted());
         }
-        return CompletableFuture.completedFuture(Response.empty().duplicate(sequenceId));
+        return CompletableFuture.completedFuture(Response.empty().duplicate(sequentialId));
       }
       log.debug("Sending commit request for transformed elements {}", transformed);
       return manager.commit(transactionId, transformed);
@@ -333,12 +333,12 @@ private void processArrivedResponse(@Nullable Response response, @Nullable Throw
       }
       if (response.hasSequenceId()) {
         Preconditions.checkState(
-            sequenceId == -1 || sequenceId == response.getSeqId(),
+            sequentialId == -1 || sequentialId == response.getSeqId(),
             "Updated sequence ID from %s to %s. That is a bug in proxima's transactions.",
-            sequenceId,
+            sequentialId,
             response.getSeqId());
-        sequenceId = response.getSeqId();
-        log.debug("Assigned sequence ID {} for transaction {}", sequenceId, transactionId);
+        sequentialId = response.getSeqId();
+        log.debug("Assigned sequence ID {} for transaction {}", sequentialId, transactionId);
       }
       if (response.hasStamp()) {
         Preconditions.checkState(
@@ -428,13 +428,13 @@ private CompletableFuture<Void> applyTransform(
     private StreamElement injectSequenceIdAndStamp(StreamElement in) {
 
       Preconditions.checkState(
-          sequenceId > 0, "Invalid sequence ID %s for %s", sequenceId, transactionId);
+          sequentialId > 0, "Invalid sequence ID %s for %s", sequentialId, transactionId);
       Preconditions.checkArgument(!in.isDeleteWildcard(), "Wildcard deletes not yet supported");
 
       return StreamElement.upsert(
           in.getEntityDescriptor(),
           in.getAttributeDescriptor(),
-          sequenceId,
+          sequentialId,
           in.getKey(),
           in.getAttribute(),
           stamp,

direct/core/src/main/java/cz/o2/proxima/direct/core/view/LocalCachedPartitionedView.java

@@ -108,13 +108,13 @@ protected void onCache(StreamElement element, boolean overwrite) {
       final Pair<Long, Payload> oldVal;
       synchronized (cache) {
         oldVal = cache.get(element.getKey(), attrName, Long.MAX_VALUE);
-        long sequenceId = element.hasSequentialId() ? element.getSequentialId() : 0L;
+        long sequentialId = element.hasSequentialId() ? element.getSequentialId() : 0L;
         updated =
             cache.put(
                 element.getKey(),
                 attrName,
                 element.getStamp(),
-                sequenceId,
+                sequentialId,
                 overwrite,
                 parsed.orElse(null));
       }

direct/core/src/main/java/cz/o2/proxima/direct/core/view/TimeBoundedVersionedCache.java

@@ -198,7 +198,7 @@ synchronized boolean put(
       String key,
       String attribute,
       long stamp,
-      long sequenceId,
+      long sequentialId,
       boolean overwrite,
       @Nullable Object value) {
 
@@ -215,7 +215,7 @@ synchronized boolean put(
             final Payload oldPayload = valueMap.get(stamp);
             if (overwrite || oldPayload == null || oldPayload.overridable) {
               logPayloadUpdateIfNecessary(key, attribute, stamp, value);
-              Payload newPayload = new Payload(value, sequenceId, !overwrite);
+              Payload newPayload = new Payload(value, sequentialId, !overwrite);
               valueMap.put(stamp, newPayload);
               updated.set(!newPayload.equals(oldPayload));
             }

direct/io-kafka/src/main/java/cz/o2/proxima/direct/io/kafka/KafkaAccessor.java

@@ -48,7 +48,6 @@
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.DescribeConfigsResult;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.config.TopicConfig;
 
@@ -75,10 +74,6 @@ public class KafkaAccessor extends SerializableAbstractStorage implements DataAc
   /** Maximal read speed in bytes per second. */
   public static final String MAX_BYTES_PER_SEC = "bytes-per-sec-max";
 
-  /** Number of records per poll() */
-  public static final String MAX_POLL_RECORDS =
-      KAFKA_CONFIG_PREFIX + ConsumerConfig.MAX_POLL_RECORDS_CONFIG;
-
   /** Auto commit interval in milliseconds. */
   public static final String AUTO_COMMIT_INTERVAL_MS = "commit.auto-interval-ms";
 
@@ -115,8 +110,6 @@ public class KafkaAccessor extends SerializableAbstractStorage implements DataAc
 
   @Getter private long maxBytesPerSec = Long.MAX_VALUE;
 
-  @Getter private int maxPollRecords = 500;
-
   @Getter private long autoCommitIntervalMs = Long.MAX_VALUE;
 
   @Getter private long logStaleCommitIntervalMs = 60_000L;
@@ -164,11 +157,6 @@ private void configure(Map<String, Object> cfg) {
             .map(v -> Long.valueOf(v.toString()))
             .orElse(maxBytesPerSec);
 
-    this.maxPollRecords =
-        Optional.ofNullable(cfg.get(MAX_POLL_RECORDS))
-            .map(v -> Integer.valueOf(v.toString()))
-            .orElse(maxPollRecords);
-
     this.autoCommitIntervalMs =
         Optional.ofNullable(cfg.get(AUTO_COMMIT_INTERVAL_MS))
             .map(v -> Long.parseLong(v.toString()))
@@ -200,7 +188,6 @@ private void configure(Map<String, Object> cfg) {
             + "partitionerClass {}, "
             + "maxBytesPerSec {}, "
             + "watermarkConfiguration {}, "
-            + "maxPollRecords {}, "
             + "autoCommitIntervalNs {}, "
             + "logStaleCommitIntervalMs {}, "
             + "serializerClass {},"
@@ -209,7 +196,6 @@ private void configure(Map<String, Object> cfg) {
         partitioner.getClass(),
         maxBytesPerSec,
         watermarkConfiguration,
-        maxPollRecords,
         autoCommitIntervalMs,
         logStaleCommitIntervalMs,
         serializerClass,