tombstone testing

README.md

@@ -1,16 +1,14 @@
 # time-is-up
 
-
 TODO
-- use GEO_DISTANCE for variable GEOHASH length
-- station-sink equivalent: elastic sink or websocket?
+- outer join trace_to_estimate and estimate_t
+- test tombstone deletes unmoved and track
+- station-sink equivalent
 - simplify messages
 - new version tile sink connector -> load through hub command
-- tombstone delete through message retention 1 hour?
+- use GEO_DISTANCE for variable GEOHASH length
 - order counting query
-- deeplearning UDF?
+- implement delete mechanism at pick-up
 
-- https://www.confluent.io/blog/importance-of-distributed-tracing-for-apache-kafka-based-applications/
-- https://www.confluent.io/blog/apache-kafka-producer-improvements-sticky-partitioner/: AWS: (AWS) m3.xlarge instances with SSD.
 - https://www.confluent.io/blog/importance-of-distributed-tracing-for-apache-kafka-based-applications/
 - https://github.com/openzipkin-contrib/brave-kafka-interceptor

ksqldb/queries.sql

@@ -79,9 +79,9 @@ INSERT INTO mover (id, lat, lon) VALUES ('thisisme', 0.91, 0.99);
 INSERT INTO mover (id, lat, lon) VALUES ('thisisme', 1.0, 1.0);
 
 -- mover + track = trace
-CREATE STREAM trace1
+CREATE STREAM trace
   WITH (VALUE_FORMAT = 'protobuf',
-        KAFKA_TOPIC = 'trace1',
+        KAFKA_TOPIC = 'trace',
         PARTITIONS = 2)
   AS SELECT
           mover.id AS mover_id,
@@ -98,12 +98,12 @@ CREATE STREAM trace1
 CREATE SINK CONNECTOR tile WITH (
   'tasks.max' = '1',
   'connector.class' = 'guru.bonacci.kafka.connect.tile38.Tile38SinkConnector',
-  'topics' = 'unmoved,trace1',
+  'topics' = 'unmoved,trace',
   'key.converter' = 'org.apache.kafka.connect.storage.StringConverter',
   'value.converter' = 'io.confluent.connect.protobuf.ProtobufConverter',
   'value.converter.schema.registry.url' = 'http://schema-registry:8081',
   'tile38.topic.unmoved' = 'SET unmoved event.ID POINT event.LATITUDE event.LONGITUDE',
-  'tile38.topic.trace1' = 'SET trace event.MOVER_ID POINT event.LAT event.LON',
+  'tile38.topic.trace' = 'SET trace event.MOVER_ID POINT event.LAT event.LON',
   'tile38.host' = 'tile38',
   'tile38.port' = 9851,
   'errors.tolerance' = 'all',
@@ -177,4 +177,4 @@ CREATE STREAM homeward
      estimate.togo_ms as togo_ms
     FROM trace_to_estimate AS trace
     INNER JOIN estimate_t as estimate ON estimate.hashkey = trace.hashkey
-    PARTITION BY trace.unmoved_id;
+    PARTITION BY trace.tracking_number;

pom.xml

@@ -39,6 +39,11 @@
 			<artifactId>commons-lang3</artifactId>
 			<version>3.10</version>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.kafka</groupId>
+			<artifactId>reactor-kafka</artifactId>
+			<version>1.2.2.RELEASE</version>
+		</dependency>
 		<dependency>
 			<groupId>com.google.protobuf</groupId>
 			<artifactId>protobuf-java</artifactId>

src/main/java/guru/bonacci/timesup/streams/MoverProducer.java

@@ -0,0 +1,86 @@
+package guru.bonacci.timesup.streams;
+
+import java.text.SimpleDateFormat;
+import java.time.Duration;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import guru.bonacci.timesup.model.TheMover.Mover;
+import reactor.core.publisher.Flux;
+import reactor.kafka.sender.KafkaSender;
+import reactor.kafka.sender.SenderOptions;
+import reactor.kafka.sender.SenderRecord;
+
+public class MoverProducer {
+
+    private static final Logger log = LoggerFactory.getLogger(MoverProducer.class.getName());
+
+    static final String BOOTSTRAP_SERVERS = "localhost:29092";
+    static final String SCHEMA_REGISTRY = "http://127.0.0.1:8081";
+    static final String TOPIC = "testmover";
+
+    private final KafkaSender<String, Mover> sender;
+    private final SimpleDateFormat dateFormat;
+
+    public MoverProducer(String bootstrapServers) {
+
+        Map<String, Object> props = new HashMap<>();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        props.put(ProducerConfig.CLIENT_ID_CONFIG, "sample-producer");
+        props.put(ProducerConfig.ACKS_CONFIG, "all");
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
+        		  "io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer");
+		props.put("schema.registry.url", SCHEMA_REGISTRY);
+		SenderOptions<String, Mover> senderOptions = SenderOptions.create(props);
+
+        sender = KafkaSender.create(senderOptions);
+        dateFormat = new SimpleDateFormat("HH:mm:ss:SSS z dd MMM yyyy");
+    }
+
+//	String loc = step.getLeft() + "," + step.getRight(); 
+    public void sendMessages(String topic, CountDownLatch latch) throws InterruptedException {
+        sender.send(Flux.interval(Duration.ofSeconds(1))
+                        .map(i -> SenderRecord.create(toRecord(topic), i)))
+              .doOnError(e -> log.error("Send failed", e))
+              .subscribe(r -> {
+                  RecordMetadata metadata = r.recordMetadata();
+                  System.out.printf("Message %d sent successfully, topic-partition=%s-%d offset=%d timestamp=%s\n",
+                      r.correlationMetadata(),
+                      metadata.topic(),
+                      metadata.partition(),
+                      metadata.offset(),
+                      dateFormat.format(new Date(metadata.timestamp())));
+                  latch.countDown();
+              });
+    }
+
+    ProducerRecord<String, Mover> toRecord(String topic) {
+    	Mover record = Mover.newBuilder().setId("foo").setLat(1.1f).setLon(1.0f).build();
+//    	record = null;
+    	return new ProducerRecord<>(topic, "foo", record);
+    }
+
+    public void close() {
+        sender.close();
+    }
+
+    public static void main(String[] args) throws Exception {
+        int count = 1;
+        CountDownLatch latch = new CountDownLatch(count);
+        MoverProducer producer = new MoverProducer(BOOTSTRAP_SERVERS);
+        producer.sendMessages(TOPIC, latch);
+        latch.await(5, TimeUnit.MINUTES);
+        producer.close();
+    }
+}

src/main/java/guru/bonacci/timesup/streams/UnmovedProducer.java

@@ -0,0 +1,86 @@
+package guru.bonacci.timesup.streams;
+
+import java.text.SimpleDateFormat;
+import java.time.Duration;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import guru.bonacci.timesup.model.TheUnmoved.Unmoved;
+import reactor.core.publisher.Flux;
+import reactor.kafka.sender.KafkaSender;
+import reactor.kafka.sender.SenderOptions;
+import reactor.kafka.sender.SenderRecord;
+
+public class UnmovedProducer {
+
+    private static final Logger log = LoggerFactory.getLogger(UnmovedProducer.class.getName());
+
+    static final String BOOTSTRAP_SERVERS = "localhost:29092";
+    static final String SCHEMA_REGISTRY = "http://127.0.0.1:8081";
+    static final String TOPIC = "testunmoved";
+
+    private final KafkaSender<String, Unmoved> sender;
+    private final SimpleDateFormat dateFormat;
+
+    public UnmovedProducer(String bootstrapServers) {
+
+        Map<String, Object> props = new HashMap<>();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        props.put(ProducerConfig.CLIENT_ID_CONFIG, "sample-producer");
+        props.put(ProducerConfig.ACKS_CONFIG, "all");
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
+        		  "io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer");
+		props.put("schema.registry.url", SCHEMA_REGISTRY);
+		SenderOptions<String, Unmoved> senderOptions = SenderOptions.create(props);
+
+        sender = KafkaSender.create(senderOptions);
+        dateFormat = new SimpleDateFormat("HH:mm:ss:SSS z dd MMM yyyy");
+    }
+
+//	String loc = step.getLeft() + "," + step.getRight(); 
+    public void sendMessages(String topic, CountDownLatch latch) throws InterruptedException {
+        sender.send(Flux.interval(Duration.ofSeconds(1))
+                        .map(i -> SenderRecord.create(toRecord(topic), i)))
+              .doOnError(e -> log.error("Send failed", e))
+              .subscribe(r -> {
+                  RecordMetadata metadata = r.recordMetadata();
+                  System.out.printf("Message %d sent successfully, topic-partition=%s-%d offset=%d timestamp=%s\n",
+                      r.correlationMetadata(),
+                      metadata.topic(),
+                      metadata.partition(),
+                      metadata.offset(),
+                      dateFormat.format(new Date(metadata.timestamp())));
+                  latch.countDown();
+              });
+    }
+
+    ProducerRecord<String, Unmoved> toRecord(String topic) {
+    	Unmoved record = Unmoved.newBuilder().setId("bar").setLatitude(1.0f).setLongitude(1.0f).build();
+//    	record = null;
+    	return new ProducerRecord<>(topic, "foo", record);
+    }
+
+    public void close() {
+        sender.close();
+    }
+
+    public static void main(String[] args) throws Exception {
+        int count = 1;
+        CountDownLatch latch = new CountDownLatch(count);
+        UnmovedProducer producer = new UnmovedProducer(BOOTSTRAP_SERVERS);
+        producer.sendMessages(TOPIC, latch);
+        latch.await(5, TimeUnit.MINUTES);
+        producer.close();
+    }
+}

src/main/protobuf/the-unmoved.proto

@@ -7,7 +7,6 @@ option java_outer_classname = "TheUnmoved";
 
 message Unmoved {
   required string id = 1;
-  optional string name = 2;
   required float latitude = 3;
   required float longitude = 4;
 }