The above may not be an issue as in it just uses the returned class loader to resolve the Connector I think . What is not obvious, why it does not go ahead and consume ..
[mm2-dev-749469cf68-vpm2l] [2019-10-24 20:26:07,571] INFO refreshing known target topics took 15 ms (org.apache.kafka.connect.mirror.Scheduler:95) [mm2-dev-749469cf68-vpm2l] [2019-10-24 20:26:07,574] INFO Started MirrorSourceConnector with 120 topic-partitions. (org.apache.kafka.connect.mirror.MirrorSourceConnector:121) [mm2-dev-749469cf68-vpm2l] [2019-10-24 20:26:07,574] INFO Starting MirrorSourceConnector took 160 ms. (org.apache.kafka.connect.mirror.MirrorSourceConnector:122) [mm2-dev-749469cf68-vpm2l] [2019-10-24 20:26:07,577] INFO Finished creating connector MirrorSourceConnector (org.apache.kafka.connect.runtime.Worker:272) [mm2-dev-749469cf68-vpm2l] [2019-10-24 20:26:07,578] ERROR Plugin class loader for connector: 'org.apache.kafka.connect.mirror.MirrorSourceConnector' was not found. Returning: org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader@2262b621 (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:165) [mm2-dev-749469cf68-vpm2l] [2019-10-24 20:26:07,579] INFO SourceConnectorConfig values: [mm2-dev-749469cf68-vpm2l] config.action.reload = restart [mm2-dev-749469cf68-vpm2l] connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector [mm2-dev-749469cf68-vpm2l] errors.log.enable = false [mm2-dev-749469cf68-vpm2l] errors.log.include.messages = false [mm2-dev-749469cf68-vpm2l] errors.retry.delay.max.ms = 60000 [mm2-dev-749469cf68-vpm2l] errors.retry.timeout = 0 [mm2-dev-749469cf68-vpm2l] errors.tolerance = none [mm2-dev-749469cf68-vpm2l] header.converter = null [mm2-dev-749469cf68-vpm2l] key.converter = null [mm2-dev-749469cf68-vpm2l] name = MirrorSourceConnector [mm2-dev-749469cf68-vpm2l] tasks.max = 48 [mm2-dev-749469cf68-vpm2l] transforms = [] [mm2-dev-749469cf68-vpm2l] value.converter = null [mm2-dev-749469cf68-vpm2l] (org.apache.kafka.connect.runtime.SourceConnectorConfig:347) [mm2-dev-749469cf68-vpm2l] [2019-10-24 20:26:07,580] INFO EnrichedConnectorConfig values: [mm2-dev-749469cf68-vpm2l] config.action.reload = restart [mm2-dev-749469cf68-vpm2l] connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector [mm2-dev-749469cf68-vpm2l] errors.log.enable = false [mm2-dev-749469cf68-vpm2l] errors.log.include.messages = false [mm2-dev-749469cf68-vpm2l] errors.retry.delay.max.ms = 60000 [mm2-dev-749469cf68-vpm2l] errors.retry.timeout = 0 [mm2-dev-749469cf68-vpm2l] errors.tolerance = none [mm2-dev-749469cf68-vpm2l] header.converter = null [mm2-dev-749469cf68-vpm2l] key.converter = null [mm2-dev-749469cf68-vpm2l] name = MirrorSourceConnector [mm2-dev-749469cf68-vpm2l] tasks.max = 48 [mm2-dev-749469cf68-vpm2l] transforms = [] [mm2-dev-749469cf68-vpm2l] value.converter = null [mm2-dev-749469cf68-vpm2l] (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig:347) And from there on nothing.. On Thu, Oct 24, 2019 at 3:02 PM Vishal Santoshi <vishal.santo...@gmail.com> wrote: > Hey Ryanne, > > Seeing the below ERROR in the logs and then, it seems the > process does not consume ( it does not exit with any errors ) . And this is > intermittent. As in do it enough times. that does relaunch :) Is this > something a known bug > > [mm2-dev-58bf5df684-ln9k2] [2019-10-24 18:41:03,067] ERROR > Plugin class loader for connector: > 'org.apache.kafka.connect.mirror.MirrorHeartbeatConnector' was not found. > Returning: > org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader@2262b621 > (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:165) > > > > > > > On Mon, Oct 21, 2019 at 5:16 PM Ryanne Dolan <ryannedo...@gmail.com> > wrote: > >> Vishal, the number of tasks created per source->target herder is >> determined >> by both tasks.max and the total number of topic-partitions being >> replicated. In order to use all 12 worker nodes, you'd need tasks.max >= >> 12 >> and number of topic-partitions >= 12. From previous emails it sounds like >> you have a small number of topic-partitions total (i.e. a small number of >> topics with a small number of partitions per topic), so I'm guessing >> that's >> the reason you aren't seeing more tasks being created. >> >> Ryanne >> >> On Sat, Oct 19, 2019 at 1:28 AM Vishal Santoshi < >> vishal.santo...@gmail.com> >> wrote: >> >> > Here is what I see >> > >> > * The max tasks are a a cap on a Connector across the cluster. If have >> 8 >> > VMs but 8 max tasks my assumption that there would be 8 * 8 = 72 task >> > threads was >> > wring. The logs showed that the partitions were consumed by 8 threads >> on >> > the 8 VMs ( 1 per VM ) which was highly un optimal. When I scaled the >> > VMs to 12, it did not matter, as the max tasks still prevented any >> further >> > distribution. >> > >> > * If I cancel/resume the cluster with a max task of 48 ( keeping the >> same >> > job name and thus connector definition the max tasks does not change, >> as >> > in >> > it seems to keep the same number of max task threads limit ( as in 8 ) >> > >> > * I can bring down a VM and see the task migrate to a free VM but the >> > overall count of task threads remain the same. >> > >> > >> > In essence, the num of tasks is a cap on threads in the cluster per >> > connector, A connector is a source->sink pair that spans a cluster. >> Thus if >> > we have a >> > A->B DAG and max tasks of 8, then there will be no more that 8 Source >> > Tasks ( threads ) no matter how big the cluster is, It thus makes >> sense to >> > over provision ( within limits of a single VM ) on the max tasks to >> allow >> > for adding more VMs for scale up..... >> > >> > >> > >> > On Fri, Oct 18, 2019 at 8:04 PM Vishal Santoshi < >> vishal.santo...@gmail.com >> > > >> > wrote: >> > >> > > I misspoke >> > > >> > > >> I now have 8 VMs 8 cpus with 48 max tasks and it did spread to the >> the >> > > 8 VMs. I then upscaled to 12 VMs and the tasks *have not *migrated >> as I >> > > would expect . >> > > >> > > >> > > >> > > >> > > On Fri, Oct 18, 2019 at 8:00 PM Vishal Santoshi < >> > vishal.santo...@gmail.com> >> > > wrote: >> > > >> > >> OK, You will have to explain :) >> > >> >> > >> I had 12 VMs with 8 cpus and 8 max tasks. I thought let me give a >> CPU >> > to >> > >> each task, which I presumed is a java thread ( even though I know the >> > >> thread would be mostly ip bound ). . I saw the issue I pointed up. >> > >> *I now have 8 VMs 8 cpus with 48 max tasks and it did spread to the >> the >> > >> 8 VMs. I then upscaled to 12 VMs and the tasks migrated as I would >> > expect >> > >> .* >> > >> >> > >> I know that a VM will have MirrorSourceConnector and >> > >> MirrorHeartbeatConnector tasks up till tasks.max. So a few >> questions >> > >> >> > >> >> > >> >> > >> * When we say there are 48 max tasks, are we saying there are 48 >> > threads >> > >> ( in fact 96, each for the 2 groups above, worst case + 2 ) ? >> > >> * When we talk about Connector, are we talking about a JVM process, >> as >> > in >> > >> a Connector is a JVM process ? >> > >> * Why larger number of tasks.max help the spread ? As in I would >> > assume >> > >> there are up till 8 tasks ( or 16 ) per VM but how that should not >> have >> > >> prevented re assignment on a scale up ( as it clearly did ) ? >> > >> >> > >> The reason I ask is that I plan to run mm2 cluster on k8s and I >> want to >> > >> make sure that I use the version of JVM that is more docker friendly >> > vis a >> > >> vis, how many cpus it believes it has and as explained here >> > >> >> > >> https://blog.softwaremill.com/docker-support-in-new-java-8-finally-fd595df0ca54 >> > >> >> > >> >> > >> >> > >> >> > >> On Fri, Oct 18, 2019 at 4:15 PM Ryanne Dolan <ryannedo...@gmail.com> >> > >> wrote: >> > >> >> > >>> What is tasks.max? Consider bumping to something like 48 if you're >> > >>> running >> > >>> on a dozen nodes. >> > >>> >> > >>> Ryanne >> > >>> >> > >>> On Fri, Oct 18, 2019, 1:43 PM Vishal Santoshi < >> > vishal.santo...@gmail.com >> > >>> > >> > >>> wrote: >> > >>> >> > >>> > Hey Ryanne, >> > >>> > >> > >>> > >> > >>> > I see a definite issue. I am doing an intense test >> and I >> > >>> bring >> > >>> > up 12 VMs ( they are 12 pods with 8 cpus each ), replicating about >> > 1200 >> > >>> > plus topics ( fairly heavy 100mbps ) ... They are acquired and are >> > >>> > staggered as they come up..I see a fraction of these nodes not >> > >>> assigned any >> > >>> > replication....There is plenty to go around. ( more then a couple >> of >> > >>> > thousand partitions ) . is there something I am missing.... As >> in >> > my >> > >>> > current case 5 of the 12 VMs are idle.. >> > >>> > >> > >>> > Vishal >> > >>> > >> > >>> > On Fri, Oct 18, 2019 at 7:05 AM Vishal Santoshi < >> > >>> vishal.santo...@gmail.com >> > >>> > > >> > >>> > wrote: >> > >>> > >> > >>> > > Oh sorry a. COUNTER... is more like it.... >> > >>> > > >> > >>> > > On Fri, Oct 18, 2019, 6:58 AM Vishal Santoshi < >> > >>> vishal.santo...@gmail.com >> > >>> > > >> > >>> > > wrote: >> > >>> > > >> > >>> > >> Will do >> > >>> > >> One more thing the age/latency metrics seem to be >> analogous as >> > >>> in >> > >>> > >> they seem to be calculated using similar routines. I would >> think >> > a >> > >>> > metric >> > >>> > >> tracking >> > >>> > >> the number of flush failures ( as a GAUGE ) given >> > >>> > >> offset.flush.timeout.ms would be highly beneficial. >> > >>> > >> >> > >>> > >> Regards.. >> > >>> > >> >> > >>> > >> >> > >>> > >> On Thu, Oct 17, 2019 at 11:53 PM Ryanne Dolan < >> > >>> ryannedo...@gmail.com> >> > >>> > >> wrote: >> > >>> > >> >> > >>> > >>> Ah, I see you are correct. Also I misspoke saying "workers" >> > >>> earlier, as >> > >>> > >>> the >> > >>> > >>> consumer is not created by the worker, but the task. >> > >>> > >>> >> > >>> > >>> I suppose the put() could be changed to putIfAbsent() here to >> > >>> enable >> > >>> > this >> > >>> > >>> property to be changed. Maybe submit a PR? >> > >>> > >>> >> > >>> > >>> Ryanne >> > >>> > >>> >> > >>> > >>> On Thu, Oct 17, 2019 at 10:00 AM Vishal Santoshi < >> > >>> > >>> vishal.santo...@gmail.com> >> > >>> > >>> wrote: >> > >>> > >>> >> > >>> > >>> > Hmm ( I did both ) >> > >>> > >>> > >> > >>> > >>> > another->another_test.enabled = true >> > >>> > >>> > >> > >>> > >>> > another->another_test.topics = act_post >> > >>> > >>> > >> > >>> > >>> > another->another_test.emit.heartbeats.enabled = false >> > >>> > >>> > >> > >>> > >>> > another->another_test.consumer.auto.offset.reset = latest >> > >>> > >>> > >> > >>> > >>> > another->another_test.sync.topic.acls.enabled = false >> > >>> > >>> > >> > >>> > >>> > another.consumer.auto.offset.reset = latest >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> > When I grep for the ConsumerConfig ( and there are 8 >> instances, >> > >>> this >> > >>> > >>> topic >> > >>> > >>> > has 4 partitions ) >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> > [2019-10-17 14:01:21,879] INFO ConsumerConfig values: >> > >>> > >>> > >> > >>> > >>> > allow.auto.create.topics = true >> > >>> > >>> > >> > >>> > >>> > auto.commit.interval.ms = 5000 >> > >>> > >>> > >> > >>> > >>> > *auto.offset.reset* = earliest >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> > I am now using the 2.4 branch from kafka trunk >> > >>> > >>> > https://github.com/apache/kafka/tree/2.4/connect/mirror >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> > This code change works and makes sense.. I think all other >> > >>> settings >> > >>> > >>> will be >> > >>> > >>> > fine ( as can be overridden ) but for the 2 below.. >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> > *--- >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorConnectorConfig.java* >> > >>> > >>> > >> > >>> > >>> > *+++ >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorConnectorConfig.java* >> > >>> > >>> > >> > >>> > >>> > @@ -230,7 +230,7 @@ public class MirrorConnectorConfig >> extends >> > >>> > >>> > AbstractConfig { >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >> > props.keySet().retainAll(MirrorClientConfig.CLIENT_CONFIG_DEF.names()); >> > >>> > >>> > >> > >>> > >>> > >> > >>> props.putAll(originalsWithPrefix(CONSUMER_CLIENT_PREFIX)); >> > >>> > >>> > >> > >>> > >>> > props.put("enable.auto.commit", "false"); >> > >>> > >>> > >> > >>> > >>> > - props.put("auto.offset.reset", "earliest"); >> > >>> > >>> > >> > >>> > >>> > + props.put("auto.offset.reset", "latest"); >> > >>> > >>> > >> > >>> > >>> > return props; >> > >>> > >>> > >> > >>> > >>> > } >> > >>> > >>> > >> > >>> > >>> > >> > >>> > >>> > Regards. >> > >>> > >>> > >> > >>> > >>> > On Wed, Oct 16, 2019 at 3:36 PM Ryanne Dolan < >> > >>> ryannedo...@gmail.com> >> > >>> > >>> > wrote: >> > >>> > >>> > >> > >>> > >>> > > Vishal, you should be able to override the properties >> passed >> > >>> to the >> > >>> > >>> > > internal workers using properties like >> > >>> > >>> A->B.consumer.auto.offset.reset or >> > >>> > >>> > > A.consumer.auto.offset.reset in the mm2.properties file. >> > >>> Certain >> > >>> > >>> > top-level >> > >>> > >>> > > properties like tasks.max are honored without the A->B or >> A >> > >>> prefix, >> > >>> > >>> but >> > >>> > >>> > > auto.offset.reset is not one of them. >> > >>> > >>> > > >> > >>> > >>> > > Ryanne >> > >>> > >>> > > >> > >>> > >>> > > On Wed, Oct 16, 2019 at 9:13 AM Vishal Santoshi < >> > >>> > >>> > vishal.santo...@gmail.com >> > >>> > >>> > > > >> > >>> > >>> > > wrote: >> > >>> > >>> > > >> > >>> > >>> > > > Hey Ryanne, >> > >>> > >>> > > > >> > >>> > >>> > > > >> > >>> > >>> > > > How do I override auto.offset.reset = latest for >> > >>> consumers >> > >>> > >>> through >> > >>> > >>> > > > mm2.properties. I have tried straight up . >> > auto.offset.reset >> > >>> and >> > >>> > >>> > > consumer. >> > >>> > >>> > > > auto.offset.reset but it defaults to earliest.. I do >> have >> > a >> > >>> > query >> > >>> > >>> in >> > >>> > >>> > > > another thread but though you might know off hand.. >> > >>> > >>> > > > >> > >>> > >>> > > > I would imagine there is some way in general of >> overriding >> > >>> > >>> consumer and >> > >>> > >>> > > > producer configs through mm2.properties in MM2 ? >> > >>> > >>> > > > >> > >>> > >>> > > > Regards. >> > >>> > >>> > > > >> > >>> > >>> > > > On Tue, Oct 15, 2019 at 3:44 PM Vishal Santoshi < >> > >>> > >>> > > vishal.santo...@gmail.com >> > >>> > >>> > > > > >> > >>> > >>> > > > wrote: >> > >>> > >>> > > > >> > >>> > >>> > > > > Thank you so much for all your help. Will keep you >> > posted >> > >>> on >> > >>> > >>> tests I >> > >>> > >>> > > > do.. >> > >>> > >>> > > > > I hope this is helpful to other folks too.. >> > >>> > >>> > > > > >> > >>> > >>> > > > > On Tue, Oct 15, 2019 at 2:44 PM Ryanne Dolan < >> > >>> > >>> ryannedo...@gmail.com> >> > >>> > >>> > > > > wrote: >> > >>> > >>> > > > > >> > >>> > >>> > > > >> That's right. MM2 is at-least-once for now, same as >> > legacy >> > >>> > >>> > > MirrorMaker. >> > >>> > >>> > > > >> You >> > >>> > >>> > > > >> can follow >> > >>> https://issues.apache.org/jira/browse/KAFKA-6080 >> > >>> > for >> > >>> > >>> > > updates >> > >>> > >>> > > > >> on >> > >>> > >>> > > > >> exactly-once semantics in Connect. >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> Ryanne >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> On Tue, Oct 15, 2019 at 1:24 PM Vishal Santoshi < >> > >>> > >>> > > > >> vishal.santo...@gmail.com> >> > >>> > >>> > > > >> wrote: >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >> You are correct. I'm working on a KIP and PoC >> to >> > >>> > >>> introduce >> > >>> > >>> > > > >> > transactions to >> > >>> > >>> > > > >> > >> Connect for this exact purpose :) >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> > That is awesome. Any time frame ? >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> > In the mean time the SLA as of now >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> > 1. It is conceivable that we flush the producer to >> the >> > >>> > target >> > >>> > >>> > > cluster >> > >>> > >>> > > > >> but >> > >>> > >>> > > > >> > fail to offset commit. If there was a restart >> before >> > the >> > >>> > next >> > >>> > >>> > > > successful >> > >>> > >>> > > > >> > offset commit, there will be duplicates and a >> part >> > of >> > >>> data >> > >>> > >>> is >> > >>> > >>> > > > >> replayed ( >> > >>> > >>> > > > >> > at least once ) ? >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> > 2. The same can be said about partial flushes, >> though >> > >>> am >> > >>> > not >> > >>> > >>> sure >> > >>> > >>> > > > about >> > >>> > >>> > > > >> > how kafka addresses flush ( Is a flush either >> success >> > >>> or a >> > >>> > >>> > failure, >> > >>> > >>> > > > and >> > >>> > >>> > > > >> > nothing in between ) >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> > Thanks.. >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> > On Tue, Oct 15, 2019 at 12:34 PM Ryanne Dolan < >> > >>> > >>> > > ryannedo...@gmail.com> >> > >>> > >>> > > > >> > wrote: >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> > > Hey Vishal, glad to hear you're making progress. >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > > > 1. It seems though that flushing [...] the >> > producer >> > >>> and >> > >>> > >>> > setting >> > >>> > >>> > > > the >> > >>> > >>> > > > >> > > > offset to the compacting topic is not atomic >> OR >> > >>> do we >> > >>> > >>> use >> > >>> > >>> > > > >> > > > transactions here ? >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > > You are correct. I'm working on a KIP and PoC to >> > >>> introduce >> > >>> > >>> > > > >> transactions >> > >>> > >>> > > > >> > to >> > >>> > >>> > > > >> > > Connect for this exact purpose :) >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > > > I think these are 4 threads ( b'coz >> num.tasks=4 ), >> > >>> and I >> > >>> > >>> have >> > >>> > >>> > 2 >> > >>> > >>> > > > >> topics >> > >>> > >>> > > > >> > > with >> > >>> > >>> > > > >> > > > 1 partition each. Do I assume this right, as in >> > >>> there >> > >>> > are >> > >>> > >>> 4 >> > >>> > >>> > > > consumer >> > >>> > >>> > > > >> > > groups >> > >>> > >>> > > > >> > > > ( on CG per thread ) ... >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > > Some details here: >> > >>> > >>> > > > >> > > - tasks.max controls the maximum number of tasks >> > >>> created >> > >>> > per >> > >>> > >>> > > > Connector >> > >>> > >>> > > > >> > > instance. Both MirrorSourceConnector and >> > >>> > >>> > MirrorCheckpointConnector >> > >>> > >>> > > > >> will >> > >>> > >>> > > > >> > > create multiple tasks (up to tasks.max), but >> > >>> > >>> > > > MirrorHeartbeatConnector >> > >>> > >>> > > > >> > only >> > >>> > >>> > > > >> > > ever creates a single task. Moreover, there >> cannot >> > be >> > >>> more >> > >>> > >>> tasks >> > >>> > >>> > > > than >> > >>> > >>> > > > >> > > topic-partitions (for MirrorSourceConnector) or >> > >>> consumer >> > >>> > >>> groups >> > >>> > >>> > > (for >> > >>> > >>> > > > >> > > MirrorCheckpointConnector). So if you have two >> > topics >> > >>> with >> > >>> > >>> one >> > >>> > >>> > > > >> partition >> > >>> > >>> > > > >> > > each and 1 consumer group total, you'll have two >> > >>> > >>> > > > MirrorSourceConnector >> > >>> > >>> > > > >> > > tasks, one MirrorHeartbeatConnector task, and one >> > >>> > >>> > > > >> > MirrorCheckpointConnector >> > >>> > >>> > > > >> > > tasks, for a total of four. And that's in one >> > >>> direction >> > >>> > >>> only: if >> > >>> > >>> > > you >> > >>> > >>> > > > >> have >> > >>> > >>> > > > >> > > multiple source->target herders enabled, each >> will >> > >>> create >> > >>> > >>> tasks >> > >>> > >>> > > > >> > > independently. >> > >>> > >>> > > > >> > > - There are no consumer groups in MM2, >> technically. >> > >>> The >> > >>> > >>> Connect >> > >>> > >>> > > > >> framework >> > >>> > >>> > > > >> > > uses the Coordinator API and internal topics to >> > divide >> > >>> > tasks >> > >>> > >>> > among >> > >>> > >>> > > > >> > workers >> > >>> > >>> > > > >> > > -- not a consumer group per se. The MM2 >> connectors >> > >>> use the >> > >>> > >>> > > assign() >> > >>> > >>> > > > >> API, >> > >>> > >>> > > > >> > > not the subscribe() API, so there are no consumer >> > >>> groups >> > >>> > >>> there >> > >>> > >>> > > > >> either. In >> > >>> > >>> > > > >> > > fact, they don't commit() either. This is nice, >> as >> > it >> > >>> > >>> > eliminates a >> > >>> > >>> > > > >> lot of >> > >>> > >>> > > > >> > > the rebalancing problems legacy MirrorMaker has >> been >> > >>> > plagued >> > >>> > >>> > with. >> > >>> > >>> > > > >> With >> > >>> > >>> > > > >> > > MM2, rebalancing only occurs when the number of >> > >>> workers >> > >>> > >>> changes >> > >>> > >>> > or >> > >>> > >>> > > > >> when >> > >>> > >>> > > > >> > the >> > >>> > >>> > > > >> > > assignments change (e.g. new topics are >> discovered). >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > > Ryanne >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > > On Tue, Oct 15, 2019 at 10:23 AM Vishal Santoshi >> < >> > >>> > >>> > > > >> > > vishal.santo...@gmail.com> >> > >>> > >>> > > > >> > > wrote: >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > > > Hey Ryanne, >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > The test was on topics that had a 7 >> day >> > >>> > >>> retention. >> > >>> > >>> > > > Which >> > >>> > >>> > > > >> > > > generally implies that the batch size for >> flush is >> > >>> > pretty >> > >>> > >>> > high ( >> > >>> > >>> > > > >> till >> > >>> > >>> > > > >> > the >> > >>> > >>> > > > >> > > > consumption becomes current ). The >> > >>> > >>> offset.flush.timeout.ms >> > >>> > >>> > > > >> defaults >> > >>> > >>> > > > >> > to >> > >>> > >>> > > > >> > > 5 >> > >>> > >>> > > > >> > > > seconds and the code will not send in the >> offsets >> > >>> if the >> > >>> > >>> flush >> > >>> > >>> > > is >> > >>> > >>> > > > >> not >> > >>> > >>> > > > >> > > > complete. Increasing that time out did solve >> the >> > >>> "not >> > >>> > >>> sending >> > >>> > >>> > > the >> > >>> > >>> > > > >> > offset >> > >>> > >>> > > > >> > > to >> > >>> > >>> > > > >> > > > topic" issue. >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > Two questions ( I am being greedy here :) ) >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > 1. It seems though that flushing the flushing >> the >> > >>> > >>> producer and >> > >>> > >>> > > > >> setting >> > >>> > >>> > > > >> > > the >> > >>> > >>> > > > >> > > > offset to the compacting topic is not atomic >> OR >> > >>> do we >> > >>> > >>> use >> > >>> > >>> > > > >> > > > transactions here ? >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > 2. I see >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> WorkerSourceTask{id=MirrorHeartbeatConnector-0} >> > >>> > flushing >> > >>> > >>> > 956435 >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > WorkerSourceTask{id=MirrorSourceConnector-1} >> > >>> flushing >> > >>> > >>> 356251 >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> WorkerSourceTask{id=MirrorCheckpointConnector-2} >> > >>> > >>> flushing 0 >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> WorkerSourceTask{id=MirrorCheckpointConnector-3} >> > >>> > >>> flushing 0 >> > >>> > >>> > > > >> > outstanding >> > >>> > >>> > > > >> > > > messages >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > I think these are 4 threads ( b'coz >> num.tasks=4 ), >> > >>> and I >> > >>> > >>> have >> > >>> > >>> > 2 >> > >>> > >>> > > > >> topics >> > >>> > >>> > > > >> > > with >> > >>> > >>> > > > >> > > > 1 partition each. Do I assume this right, as in >> > >>> there >> > >>> > are >> > >>> > >>> 4 >> > >>> > >>> > > > consumer >> > >>> > >>> > > > >> > > groups >> > >>> > >>> > > > >> > > > ( on CG per thread ) ... >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > THANKS A LOT >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > Vishal. >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > On Mon, Oct 14, 2019 at 3:42 PM Ryanne Dolan < >> > >>> > >>> > > > ryannedo...@gmail.com >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> > > > wrote: >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > > > > timed out >> > >>> > >>> > > > >> > > > > while waiting for producer to flush >> outstanding >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > > Yeah, that's what I'd expect to see if >> Connect >> > was >> > >>> > >>> unable to >> > >>> > >>> > > > send >> > >>> > >>> > > > >> > > records >> > >>> > >>> > > > >> > > > > to the downstream remote topics, e.g. if >> > >>> > >>> > min.in-sync.replicas >> > >>> > >>> > > > were >> > >>> > >>> > > > >> > > > > misconfigured. Given some data seems to >> arrive, >> > >>> it's >> > >>> > >>> > possible >> > >>> > >>> > > > that >> > >>> > >>> > > > >> > > > > everything is configured correctly but with >> too >> > >>> much >> > >>> > >>> latency >> > >>> > >>> > > to >> > >>> > >>> > > > >> > > > > successfully commit within the default >> timeouts. >> > >>> You >> > >>> > may >> > >>> > >>> > want >> > >>> > >>> > > to >> > >>> > >>> > > > >> > > increase >> > >>> > >>> > > > >> > > > > the number of tasks substantially to achieve >> > more >> > >>> > >>> > parallelism >> > >>> > >>> > > > and >> > >>> > >>> > > > >> > > > > throughput. >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > > Ryanne >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > > On Mon, Oct 14, 2019, 2:30 PM Vishal >> Santoshi < >> > >>> > >>> > > > >> > > vishal.santo...@gmail.com >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > > wrote: >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > > > Aah no.. this is more to it. Note sure if >> > >>> related >> > >>> > to >> > >>> > >>> the >> > >>> > >>> > > > above. >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://github.com/axbaretto/kafka/blob/master/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java#L114 >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > Is timing out based on >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://github.com/axbaretto/kafka/blob/master/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java#L133 >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > [2019-10-14 18:55:20,820] ERROR >> > >>> > >>> > > > >> > > > > > >> WorkerSourceTask{id=MirrorSourceConnector-0} >> > >>> Failed >> > >>> > to >> > >>> > >>> > > flush, >> > >>> > >>> > > > >> timed >> > >>> > >>> > > > >> > > out >> > >>> > >>> > > > >> > > > > > while waiting for producer to flush >> > outstanding >> > >>> > 36478 >> > >>> > >>> > > messages >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> (org.apache.kafka.connect.runtime.WorkerSourceTask:423) >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > On Mon, Oct 14, 2019 at 3:15 PM Vishal >> > Santoshi >> > >>> < >> > >>> > >>> > > > >> > > > > vishal.santo...@gmail.com >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > wrote: >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > > I think this might be it.. Could you >> > confirm. >> > >>> It >> > >>> > >>> seems >> > >>> > >>> > to >> > >>> > >>> > > be >> > >>> > >>> > > > >> on >> > >>> > >>> > > > >> > the >> > >>> > >>> > > > >> > > > > path >> > >>> > >>> > > > >> > > > > > > to commit the offsets.. but not sure... >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > [2019-10-14 15:29:14,531] ERROR Scheduler >> > for >> > >>> > >>> > > > >> > MirrorSourceConnector >> > >>> > >>> > > > >> > > > > > caught >> > >>> > >>> > > > >> > > > > > > exception in scheduled task: syncing >> topic >> > >>> ACLs >> > >>> > >>> > > > >> > > > > > > >> > >>> (org.apache.kafka.connect.mirror.Scheduler:102) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > java.util.concurrent.ExecutionException: >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > org.apache.kafka.common.errors.SecurityDisabledException: >> > >>> > >>> > > No >> > >>> > >>> > > > >> > > > Authorizer >> > >>> > >>> > > > >> > > > > > is >> > >>> > >>> > > > >> > > > > > > configured on the broker >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> org.apache.kafka.connect.mirror.MirrorSourceConnector.listTopicAclBindings(MirrorSourceConnector.java:273) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> org.apache.kafka.connect.mirror.MirrorSourceConnector.syncTopicAcls(MirrorSourceConnector.java:214) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> >> > >>> > org.apache.kafka.connect.mirror.Scheduler.run(Scheduler.java:93) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> org.apache.kafka.connect.mirror.Scheduler.executeThread(Scheduler.java:112) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> org.apache.kafka.connect.mirror.Scheduler.lambda$scheduleRepeating$0(Scheduler.java:50) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> >> > >>> > >>> > > >> > >>> > >>> >> > >>> >> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> >> > >>> > java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > at >> > >>> java.lang.Thread.run(Thread.java:748) >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > Caused by: >> > >>> > >>> > > > >> > > >> > >>> org.apache.kafka.common.errors.SecurityDisabledException: >> > >>> > >>> > > > >> > > > No >> > >>> > >>> > > > >> > > > > > > Authorizer is configured on the broker >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > > On Mon, Oct 14, 2019 at 12:30 PM Ryanne >> > Dolan >> > >>> < >> > >>> > >>> > > > >> > > ryannedo...@gmail.com >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > > > > wrote: >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > I do not have a single record in the >> > >>> offsets >> > >>> > >>> topic >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> That's definitely not normal. You are >> > correct >> > >>> > that >> > >>> > >>> > > without >> > >>> > >>> > > > >> > records >> > >>> > >>> > > > >> > > > in >> > >>> > >>> > > > >> > > > > > that >> > >>> > >>> > > > >> > > > > > >> topic, MM2 will restart from EARLIEST. >> The >> > >>> > offsets >> > >>> > >>> > should >> > >>> > >>> > > > be >> > >>> > >>> > > > >> > > stored >> > >>> > >>> > > > >> > > > > > >> periodically and whenever the connectors >> > >>> > gracefully >> > >>> > >>> > > > shutdown >> > >>> > >>> > > > >> or >> > >>> > >>> > > > >> > > > > restart. >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> Is it possible the topics don't have >> > required >> > >>> > ACLs >> > >>> > >>> or >> > >>> > >>> > > > >> something? >> > >>> > >>> > > > >> > > > Also >> > >>> > >>> > > > >> > > > > > >> note: >> > >>> > >>> > > > >> > > > > > >> Connect wants the offsets topic to have >> a >> > >>> large >> > >>> > >>> number >> > >>> > >>> > of >> > >>> > >>> > > > >> > > partitions >> > >>> > >>> > > > >> > > > > and >> > >>> > >>> > > > >> > > > > > >> to >> > >>> > >>> > > > >> > > > > > >> be compacted. Though I can't imagine >> either >> > >>> would >> > >>> > >>> > prevent >> > >>> > >>> > > > >> > commits >> > >>> > >>> > > > >> > > > from >> > >>> > >>> > > > >> > > > > > >> being sent. >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> Ryanne >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> On Mon, Oct 14, 2019 at 10:46 AM Vishal >> > >>> Santoshi >> > >>> > < >> > >>> > >>> > > > >> > > > > > >> vishal.santo...@gmail.com> >> > >>> > >>> > > > >> > > > > > >> wrote: >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> > 2nd/restore issue ( I think I need to >> > >>> solve >> > >>> > the >> > >>> > >>> > > offsets >> > >>> > >>> > > > >> topic >> > >>> > >>> > > > >> > > > issue >> > >>> > >>> > > > >> > > > > > >> > before I go with the scale up and down >> > >>> issue ) >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > As you had indicated, I went ahead and >> > >>> created >> > >>> > >>> the >> > >>> > >>> > > > offsets >> > >>> > >>> > > > >> > > topic. >> > >>> > >>> > > > >> > > > > The >> > >>> > >>> > > > >> > > > > > >> > status of the cluster ( destination >> ) is >> > >>> thus >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > opic# Partitions# BrokersBrokers >> Spread >> > >>> > %Brokers >> > >>> > >>> Skew >> > >>> > >>> > > > >> %Brokers >> > >>> > >>> > > > >> > > > > Leader >> > >>> > >>> > > > >> > > > > > >> > Skew %# ReplicasUnder Replicated >> %Leader >> > >>> > >>> SizeProducer >> > >>> > >>> > > > >> > > > > > Message/SecSummed >> > >>> > >>> > > > >> > > > > > >> > Recent Offsets >> > >>> > >>> > > > >> > > > > > >> > s8k.checkpoints.internal >> > >>> > >>> > > > >> > > > > > >> > < >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://kafka-manager.bf2.tumblr.net/clusters/grete_test/topics/s8k.checkpoints.internal >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > 1 3 60 0 0 3 0 0.00 0 >> > >>> > >>> > > > >> > > > > > >> > s8k.act_search_page >> > >>> > >>> > > > >> > > > > > >> > < >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://kafka-manager.bf2.tumblr.net/clusters/grete_test/topics/s8k.act_search_page >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > 1 3 60 0 0 3 0 6675.30 4,166,842 >> > >>> > >>> > > > >> > > > > > >> > s8k.act_reach >> > >>> > >>> > > > >> > > > > > >> > < >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://kafka-manager.bf2.tumblr.net/clusters/grete_test/topics/s8k.act_reach >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > 1 3 60 0 0 3 0 20657.92 11,579,529 >> > >>> > >>> > > > >> > > > > > >> > mm2-status.s8k.internal >> > >>> > >>> > > > >> > > > > > >> > < >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://kafka-manager.bf2.tumblr.net/clusters/grete_test/topics/mm2-status.s8k.internal >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > 5 5 100 0 0 3 0 0.00 10 >> > >>> > >>> > > > >> > > > > > >> > mm2-offsets.s8k_test.internal >> > >>> > >>> > > > >> > > > > > >> > < >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://kafka-manager.bf2.tumblr.net/clusters/grete_test/topics/mm2-offsets.s8k_test.internal >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > 1 3 60 0 0 3 0 0.00 0 >> > >>> > >>> > > > >> > > > > > >> > mm2-offsets.s8k.internal >> > >>> > >>> > > > >> > > > > > >> > < >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://kafka-manager.bf2.tumblr.net/clusters/grete_test/topics/mm2-offsets.s8k.internal >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > 1 3 60 0 0 3 0 0.00 0 >> > >>> > >>> > > > >> > > > > > >> > mm2-configs.s8k.internal >> > >>> > >>> > > > >> > > > > > >> > < >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://kafka-manager.bf2.tumblr.net/clusters/grete_test/topics/mm2-configs.s8k.internal >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > 1 3 60 0 0 3 0 0.00 13 >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > You can see . that we have the 5 ( I >> > >>> created >> > >>> > >>> bot the >> > >>> > >>> > > > >> offsets, >> > >>> > >>> > > > >> > > to >> > >>> > >>> > > > >> > > > be >> > >>> > >>> > > > >> > > > > > >> safe >> > >>> > >>> > > > >> > > > > > >> > for the below ) >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *clusters = s8k, s8k_test* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *s8k.bootstrap.servers = .....* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *s8k_test.bootstrap.servers = ......* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *# only allow replication dr1 -> dr2* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *s8k->s8k_test.enabled = true* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *s8k->s8k_test.topics = >> > >>> > >>> act_search_page|act_reach* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> *s8k->s8k_test.emit.heartbeats.enabled = >> > >>> false* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *s8k_test->s8k.enabled = false* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> *s8k_test->s8k.emit.heartbeats.enabled = >> > >>> false* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *s8k_test.replication.factor = 3* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *s8k.replication.factor = 3* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *offsets.storage.replication.factor = >> 3* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *replication.factor = 3* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *replication.policy.separator = .* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > *tasks.max = 4* >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > What seems strange is that I do not >> have >> > a >> > >>> > single >> > >>> > >>> > > record >> > >>> > >>> > > > in >> > >>> > >>> > > > >> > the >> > >>> > >>> > > > >> > > > > > offsets >> > >>> > >>> > > > >> > > > > > >> > topic.. Is that normal ? I would >> > imagine >> > >>> that >> > >>> > >>> > > without a >> > >>> > >>> > > > >> > > record, >> > >>> > >>> > > > >> > > > > > there >> > >>> > >>> > > > >> > > > > > >> is >> > >>> > >>> > > > >> > > > > > >> > no way that a restore would happen.... >> > And >> > >>> that >> > >>> > >>> is >> > >>> > >>> > > > obvious >> > >>> > >>> > > > >> > when >> > >>> > >>> > > > >> > > I >> > >>> > >>> > > > >> > > > > > >> restart >> > >>> > >>> > > > >> > > > > > >> > the mm2 instance... Find the >> screenshot >> > >>> > >>> attached. In >> > >>> > >>> > > > >> essence >> > >>> > >>> > > > >> > the >> > >>> > >>> > > > >> > > > > > latency >> > >>> > >>> > > > >> > > > > > >> > avg lag is reset \when the mm2 >> instance >> > is >> > >>> > reset >> > >>> > >>> > > > >> indicating no >> > >>> > >>> > > > >> > > > > restore >> > >>> > >>> > > > >> > > > > > >> but >> > >>> > >>> > > > >> > > > > > >> > restart from EARLIEST... I must be >> > missing >> > >>> some >> > >>> > >>> thing >> > >>> > >>> > > > >> simple >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> > On Sun, Oct 13, 2019 at 7:41 PM Ryanne >> > >>> Dolan < >> > >>> > >>> > > > >> > > > ryannedo...@gmail.com >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > > >> > wrote: >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> >> Vishal, the first issue is easy: you >> > must >> > >>> set >> > >>> > >>> > > tasks.max >> > >>> > >>> > > > to >> > >>> > >>> > > > >> > > > > something >> > >>> > >>> > > > >> > > > > > >> above >> > >>> > >>> > > > >> > > > > > >> >> 1 (the default) in order to achieve >> any >> > >>> > >>> parallelism. >> > >>> > >>> > > > This >> > >>> > >>> > > > >> > > > property >> > >>> > >>> > > > >> > > > > is >> > >>> > >>> > > > >> > > > > > >> >> passed along to the internal Connect >> > >>> workers. >> > >>> > >>> It's >> > >>> > >>> > > > >> > unfortunate >> > >>> > >>> > > > >> > > > that >> > >>> > >>> > > > >> > > > > > >> >> Connect >> > >>> > >>> > > > >> > > > > > >> >> is not smart enough to default this >> > >>> property >> > >>> > to >> > >>> > >>> the >> > >>> > >>> > > > >> number of >> > >>> > >>> > > > >> > > > > > workers. >> > >>> > >>> > > > >> > > > > > >> I >> > >>> > >>> > > > >> > > > > > >> >> suspect that will improve before >> long. >> > >>> > >>> > > > >> > > > > > >> >> >> > >>> > >>> > > > >> > > > > > >> >> For the second issue, is it possible >> you >> > >>> are >> > >>> > >>> missing >> > >>> > >>> > > the >> > >>> > >>> > > > >> > > offsets >> > >>> > >>> > > > >> > > > > > >> topic? It >> > >>> > >>> > > > >> > > > > > >> >> should exist alongside the config and >> > >>> status >> > >>> > >>> topics. >> > >>> > >>> > > > >> Connect >> > >>> > >>> > > > >> > > > should >> > >>> > >>> > > > >> > > > > > >> create >> > >>> > >>> > > > >> > > > > > >> >> this topic, but there are various >> > reasons >> > >>> this >> > >>> > >>> can >> > >>> > >>> > > fail, >> > >>> > >>> > > > >> e.g. >> > >>> > >>> > > > >> > > if >> > >>> > >>> > > > >> > > > > the >> > >>> > >>> > > > >> > > > > > >> >> replication factor is misconfigured. >> You >> > >>> can >> > >>> > try >> > >>> > >>> > > > creating >> > >>> > >>> > > > >> > this >> > >>> > >>> > > > >> > > > > topic >> > >>> > >>> > > > >> > > > > > >> >> manually or changing >> > >>> > >>> > > offsets.storage.replication.factor. >> > >>> > >>> > > > >> > > > > > >> >> >> > >>> > >>> > > > >> > > > > > >> >> Ryanne >> > >>> > >>> > > > >> > > > > > >> >> >> > >>> > >>> > > > >> > > > > > >> >> On Sun, Oct 13, 2019, 5:13 PM Vishal >> > >>> Santoshi >> > >>> > < >> > >>> > >>> > > > >> > > > > > >> vishal.santo...@gmail.com> >> > >>> > >>> > > > >> > > > > > >> >> wrote: >> > >>> > >>> > > > >> > > > > > >> >> >> > >>> > >>> > > > >> > > > > > >> >> > Using >> > >>> > >>> > > > >> > > > >> > >>> > https://github.com/apache/kafka/tree/trunk/connect/mirror >> > >>> > >>> > > > >> > > > > > as a >> > >>> > >>> > > > >> > > > > > >> >> > guide, >> > >>> > >>> > > > >> > > > > > >> >> > I have build from source the >> > >>> origin/KIP-382 >> > >>> > of >> > >>> > >>> > > > >> > > > > > >> >> > >> https://github.com/apache/kafka.git. >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > I am seeing 2 issues >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > * I brought up 2 processes on 2 >> > >>> different >> > >>> > >>> nodes ( >> > >>> > >>> > > they >> > >>> > >>> > > > >> are >> > >>> > >>> > > > >> > > > > actually >> > >>> > >>> > > > >> > > > > > >> >> pods on >> > >>> > >>> > > > >> > > > > > >> >> > k8s but that should not matter ). >> They >> > >>> share >> > >>> > >>> the >> > >>> > >>> > > > >> > > mm2.properties >> > >>> > >>> > > > >> > > > > > file >> > >>> > >>> > > > >> > > > > > >> and >> > >>> > >>> > > > >> > > > > > >> >> > are replicating ( 1-way ) 3 topics >> > with >> > >>> 8 >> > >>> > >>> > partitions >> > >>> > >>> > > > in >> > >>> > >>> > > > >> > > total. >> > >>> > >>> > > > >> > > > > > That >> > >>> > >>> > > > >> > > > > > >> >> seems >> > >>> > >>> > > > >> > > > > > >> >> > to be the way to create a >> standalone >> > mm2 >> > >>> > >>> cluster. >> > >>> > >>> > I >> > >>> > >>> > > do >> > >>> > >>> > > > >> not >> > >>> > >>> > > > >> > > > > however >> > >>> > >>> > > > >> > > > > > >> see( >> > >>> > >>> > > > >> > > > > > >> >> at >> > >>> > >>> > > > >> > > > > > >> >> > least the mbeans do not show ) any >> > >>> attempt >> > >>> > to >> > >>> > >>> > > > rebalance. >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> > >>> >> > >> https://github.com/apache/kafka/tree/trunk/connect/mirror#monitoring-an-mm2-process >> > >>> > >>> > > > >> > > > > > >> >> > mbeans >> > >>> > >>> > > > >> > > > > > >> >> > are all on a single node >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > * I restart the processes on the 2 >> > >>> nodes ( >> > >>> > >>> hard >> > >>> > >>> > stop >> > >>> > >>> > > > ans >> > >>> > >>> > > > >> > > start >> > >>> > >>> > > > >> > > > ). >> > >>> > >>> > > > >> > > > > > The >> > >>> > >>> > > > >> > > > > > >> >> > offsets for replication seem to be >> > >>> reset to >> > >>> > >>> the >> > >>> > >>> > > > >> earliest, >> > >>> > >>> > > > >> > as >> > >>> > >>> > > > >> > > if >> > >>> > >>> > > > >> > > > > it >> > >>> > >>> > > > >> > > > > > >> is a >> > >>> > >>> > > > >> > > > > > >> >> > brand new mirroring. It is also >> > obvious >> > >>> from >> > >>> > >>> the >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > "record-age-ms-avg|replication-latency-ms-avg" >> > >>> > >>> > > > >> > > > > > >> >> > which I track through the restart. >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > This implies that >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > 1. Load balancing by rebalancing is >> > not >> > >>> > >>> working. I >> > >>> > >>> > > > >> cannot >> > >>> > >>> > > > >> > > scale >> > >>> > >>> > > > >> > > > > up >> > >>> > >>> > > > >> > > > > > or >> > >>> > >>> > > > >> > > > > > >> >> down >> > >>> > >>> > > > >> > > > > > >> >> > by adding nodes to the mm2 cluster >> or >> > >>> > removing >> > >>> > >>> > them. >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > 2. Restore on a mirror is not >> working. >> > >>> If >> > >>> > the >> > >>> > >>> MM2 >> > >>> > >>> > > > >> cluster >> > >>> > >>> > > > >> > is >> > >>> > >>> > > > >> > > > > > brought >> > >>> > >>> > > > >> > > > > > >> >> down, >> > >>> > >>> > > > >> > > > > > >> >> > it does not start mirroring from >> the >> > >>> last >> > >>> > >>> known >> > >>> > >>> > > > state. I >> > >>> > >>> > > > >> > see >> > >>> > >>> > > > >> > > > the, >> > >>> > >>> > > > >> > > > > > >> >> > state/config topics etc created as >> > >>> > expected.. >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > The mm2.properties is pretty >> mimimal >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > *clusters = a , b* >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > *a.bootstrap.servers = k.....* >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > *b.bootstrap.servers = k.....* >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > *# only allow replication dr1 -> >> dr2* >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > *a->b.enabled = true* >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > *a->b.topics = act_search_page* >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > *a->b.emit.heartbeats.enabled = >> false* >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > *b->a..enabled = false* >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > *b->a.emit.heartbeats.enabled = >> false* >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > What do you think is the issue ? >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> > Thanks >> > >>> > >>> > > > >> > > > > > >> >> > >> > >>> > >>> > > > >> > > > > > >> >> >> > >>> > >>> > > > >> > > > > > >> > >> > >>> > >>> > > > >> > > > > > >> >> > >>> > >>> > > > >> > > > > > > >> > >>> > >>> > > > >> > > > > > >> > >>> > >>> > > > >> > > > > >> > >>> > >>> > > > >> > > > >> > >>> > >>> > > > >> > > >> > >>> > >>> > > > >> > >> > >>> > >>> > > > >> >> > >>> > >>> > > > > >> > >>> > >>> > > > >> > >>> > >>> > > >> > >>> > >>> > >> > >>> > >>> >> > >>> > >> >> > >>> > >> > >>> >> > >> >> > >> >
