iundarigun

@Configuration
class EntityManagerConfiguration(
    localContainerEntityManagerFactoryBean: LocalContainerEntityManagerFactoryBean
) {
    init {
        if (localContainerEntityManagerFactoryBean.jpaDialect is HibernateJpaDialect) {
            (localContainerEntityManagerFactoryBean.jpaDialect as HibernateJpaDialect)
                .setPrepareConnection(false)
        }
    }

iundarigun

class TenantDataSource : AbstractRoutingDataSource() {
    companion object {
        val ctx = ThreadLocal<String>()

        fun setClient(client: String) {
            ctx.set(client)
        }
    }

    override fun determineCurrentLookupKey(): Any? {

iundarigun

@Configuration
class TransactionRoutingConfiguration {
    @Bean
    @Primary
    fun dataSource(): DataSource {
        val transactionRoutingDataSource = TenantDataSource()
        
        val datasource1 = buildDataSourceFromProperties("datasource1")
        val datasource2 = buildDataSourceFromProperties("datasource2")
        val targetDataSources: MutableMap<Any, Any> = mutableMapOf(

iundarigun

FROM openjdk:8-jre-alpine

RUN wget https://downloads.apache.org/kafka/2.5.0/kafka_2.12-2.5.0.tgz && \
    tar -xvzf kafka_2.12-*.tgz && \
    mv kafka*/ /opt/kafka

RUN apk update &&  \
    apk add bash

COPY *.properties /opt/mirror-maker/

iundarigun

class MirroringMessageHandler(
    private val arg: String = ""
) : MirrorMaker.MirrorMakerMessageHandler {

    override fun handle(record: BaseConsumerRecord?):
            MutableList<ProducerRecord<ByteArray, ByteArray>> {
        return if (record != null && mustMirroring(record)) {
            mutableListOf(
                ProducerRecord<ByteArray, ByteArray>(
                    record.topic(),

iundarigun

public class RenameTopicHandler implements MirrorMaker.MirrorMakerMessageHandler {
  private final HashMap<String, String> topicMap = new HashMap<String, String>();
  private long NO_TIMESTAMP = -1L;

  public RenameTopicHandler(String topics) {
    if(topics != null && !topics.isEmpty()){
      for (String topicAssignment : topics.split(";")) {
        String[] topicsArray = topicAssignment.split(",");
        if (topicsArray.length == 2) {
          topicMap.put(topicsArray[0], topicsArray[1]);

iundarigun

# ...
bootstrap.servers=localhost:9096

# specify the compression codec for all data generated: none, gzip, snappy, lz4, zstd
compression.type=gzip

max.in.flight.requests.per.connection=5

# other properties

iundarigun

# ...
bootstrap.servers=localhost:9094

# consumer group id
group.id=devcave-mirrormaker-group

# What to do when there is no initial offset in Kafka or if the current
# offset does not exist any more on the server: latest, earliest, none
auto.offset.reset=latest

iundarigun

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software

iundarigun

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
# 
#    http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software