remove unused code, remove usage of deprecated API's

Author: flyaruu

floodplain-dsl/src/main/kotlin/io/floodplain/kotlindsl/Floodplain.kt

@@ -44,12 +44,9 @@ import io.floodplain.streams.api.Topic
 import io.floodplain.streams.api.TopologyContext
 import io.floodplain.streams.remotejoin.TopologyConstructor
 import org.apache.kafka.connect.sink.SinkConnector
-import org.apache.kafka.connect.sink.SinkRecord
 import org.apache.kafka.connect.sink.SinkTask
 import java.time.Duration
 import java.util.Optional
-import java.util.concurrent.atomic.AtomicLong
-import kotlin.system.measureTimeMillis
 
 private val logger = mu.KotlinLogging.logger {}
 

floodplain-dsl/src/main/kotlin/io/floodplain/kotlindsl/LocalRuntime.kt

@@ -175,7 +175,6 @@ class LocalDriverContext(
 
     val connectJobs = mutableListOf<Job>()
     private val inputTopics = mutableMapOf<String, TestInputTopic<String, ReplicationMessage>>()
-    // private val rawInputTopics = mutableMapOf<String, TestInputTopic<String, ByteArray>>()
 
     private val outputTopics = mutableMapOf<String, TestOutputTopic<String, ReplicationMessage>>()
 
@@ -277,8 +276,8 @@ class LocalDriverContext(
         val topics = topics()
         val deserializer = JsonDeserializer() // TODO protobuf issue, topic is not in json connect
         val mapper = ObjectMapper()
-        mapper.findAndRegisterModules()
-        val fallback = FallbackReplicationMessageParser()
+        mapper.findAndRegisterModules() // use shared mapping object
+        val fallback = FallbackReplicationMessageParser() // use shared parser
         val sourceFlow = outputFlowSingle()
             .map { (topic, key, value) ->
                 val parsed = if (value == null) null else fallback.parseBytes(Optional.ofNullable(topic.qualifiedString()), value).valueMap(

floodplain-dsl/src/main/kotlin/io/floodplain/kotlindsl/Stream.kt

@@ -304,40 +304,17 @@ class Stream(override val topologyContext: TopologyContext, val topologyConstruc
             connectorClientConfigOverridePolicy
         )
 
-        val statusBackingStore: StatusBackingStore = KafkaStatusBackingStore(time, worker.getInternalValueConverter())
+        val statusBackingStore: StatusBackingStore = KafkaStatusBackingStore(time, worker.internalValueConverter)
         statusBackingStore.configure(config)
 
-        val configBackingStore = KafkaConfigBackingStore(worker.getInternalValueConverter(), config, worker.configTransformer())
+        val configBackingStore = KafkaConfigBackingStore(worker.internalValueConverter, config, worker.configTransformer())
         val herder = DistributedHerder(config, time, worker, kafkaClusterId, statusBackingStore, configBackingStore, advertisedUrl.toString(), connectorClientConfigOverridePolicy)
 
-        // val herder: Herder = StandaloneHerder(worker, kafkaClusterId, connectorClientConfigOverridePolicy)
         val connect = Connect(herder, rest)
 
         connect.start()
         logger.info("Connect started!!")
         return herder
-        // try {
-        //     // for (connectorPropsFile in Arrays.copyOfRange(args, 1, args.length)) {
-        //     //     val connectorProps: Map<String, String> = Utils.propsToStringMap(Utils.loadProps(connectorPropsFile))
-        //     //     val cb: FutureCallback<Herder.Created<ConnectorInfo>> =
-        //     //         FutureCallback(object : Callback<Herder.Created<ConnectorInfo?>?>() {
-        //     //             fun onCompletion(error: Throwable?, info: Herder.Created<ConnectorInfo>) {
-        //     //                 if (error != null) log.error(
-        //     //                     "Failed to create job for {}",
-        //     //                     connectorPropsFile
-        //     //                 ) else log.info("Created connector {}", info.result().name())
-        //     //             }
-        //     //         })
-        //     //     herder.putConnectorConfig(
-        //     //         connectorProps[ConnectorConfig.NAME_CONFIG],
-        //     //         connectorProps, false, cb
-        //     //     )
-        //     //     cb.get()
-        //     // }
-        // } catch (t: Throwable) {
-        //     logger.error("Stopping after connector error", t)
-        //     connect.stop()
-        // }
     }
 
     /**

floodplain-elasticsearch/src/main/kotlin/io/floodplain/elasticsearch/ElasticSearchSink.kt

@@ -68,7 +68,7 @@ fun PartialStream.elasticSearchSink(sinkName: String, topicName: String, config:
         "tasks.max" to "1",
         "type.name" to "_doc",
         "value.converter" to "org.apache.kafka.connect.json.JsonConverter",
-        "key.converter" to "org.apache.kafka.connect.json.JsonConverter", // maps not supported by elasticsearch
+        "key.converter" to "org.apache.kafka.connect.storage.StringConverter", // maps not supported by elasticsearch
         "topics" to topic.qualifiedString(),
         "schema.ignore" to "true",
         "behavior.on.null.values" to "delete",

floodplain-example/src/main/kotlin/io/floodplain/kotlindsl/example/Dispensation.kt

@@ -22,7 +22,6 @@
 import io.floodplain.immutable.factory.ImmutableFactory;
 import io.floodplain.replication.api.ReplicationMessage;
 import io.floodplain.replication.factory.ReplicationFactory;
-import org.apache.kafka.streams.processor.AbstractProcessor;
 import org.apache.kafka.streams.processor.api.Processor;
 import org.apache.kafka.streams.processor.api.ProcessorContext;
 import org.apache.kafka.streams.processor.api.Record;
@@ -50,7 +49,7 @@ public void process(Record<String, ReplicationMessage> record) {
         }
         if(record.value().operation()!= ReplicationMessage.Operation.DELETE) {
             ImmutableMessage applied = function.apply(record.key(), record.value().message(), record.value().paramMessage().orElse(ImmutableFactory.empty()));
-            context.forward(new Record<String, ReplicationMessage>(record.key(), ReplicationFactory.standardMessage(applied), record.timestamp())); //.withParamMessage(value.paramMessage()); //.orElse(ImmutableFactory.empty())).withOperation(operation));
+            context.forward(new Record<>(record.key(), ReplicationFactory.standardMessage(applied), record.timestamp())); //.withParamMessage(value.paramMessage()); //.orElse(ImmutableFactory.empty())).withOperation(operation));
 
         } else {
             // Skip processing, just forward message if it is a delete

floodplain-stream-topology/src/main/java/io/floodplain/reactive/source/topology/FunctionProcessor.java

@@ -57,7 +57,6 @@ class InstantiatedKafkaContainer(customizer: ((KafkaContainer) -> KafkaContainer
         .apply { withStartupTimeout(Duration.ofMinutes(5)) }
         .apply { withStartupAttempts(50) }
         .apply { customizer?.invoke(this) }
-    // .apply { withEnv(env) }
     var host: String
     var exposedPort: Int = -1
     init {

floodplain-test/src/main/kotlin/io/floodplain/test/InstantiatedContainer.kt

@@ -71,8 +71,6 @@ enum ValueType {
 
     ImmutableMessage merge(ImmutableMessage other, Optional<List<String>> only);
 
-    String toFlatString(ImmutableMessageParser parser);
-
     Optional<List<ImmutableMessage>> subMessages(String field);
 
     Optional<ImmutableMessage> subMessage(String field);

immutable-api/src/main/java/io/floodplain/immutable/api/ImmutableMessage.java

@@ -19,15 +19,10 @@
 package io.floodplain.immutable.impl;
 
 import io.floodplain.immutable.api.ImmutableMessage;
-import io.floodplain.immutable.api.ImmutableMessageParser;
 import io.floodplain.immutable.factory.ImmutableFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.ZoneOffset;
 import java.util.*;
 import java.util.Map.Entry;
 import java.util.function.Function;
@@ -109,19 +104,6 @@ public Map<String, Object> valueMap(boolean ignoreNull, Set<String> ignore, List
         return Collections.unmodifiableMap(result);
     }
 
-//    private Object postProcessValue(Object value) {
-//        if(value instanceof LocalDate) {
-//            return ((LocalDate)value).toEpochDay();
-//        }
-//        if(value instanceof LocalDateTime) {
-//            return ((LocalDateTime)value).toInstant(ZoneOffset.UTC).toEpochMilli();
-//        }
-//        if (value instanceof LocalTime) {
-//            return ((LocalTime)value).toSecondOfDay();
-//        }
-//        return value;
-//    }
-
     @Override
     public Set<String> columnNames() {
         return this.values.keySet();
@@ -250,17 +232,6 @@ public ImmutableMessage with(String key, Object value, ValueType type) {
         }
     }
 
-
-    @Override
-    public String toFlatString(ImmutableMessageParser parser) {
-        if (parser == null) {
-            logger.info("Can not flatten parser, no parser present");
-            return "";
-        } else {
-            return parser.describe(this);
-        }
-    }
-
     @Override
     public ImmutableMessage withoutSubMessages(String field) {
         Map<String, List<ImmutableMessage>> res = new HashMap<>(this.subMessagesMap);

immutable-impl/src/main/java/io/floodplain/immutable/impl/ImmutableMessageImpl.java

@@ -36,8 +36,6 @@ public class TestConverter {
 
     @Before
     public void setup() {
-//		ReplicationFactory.setInstance(new FallbackReplicationMessageParser());
-
     }
 
     @Test