You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "Klose6 (via GitHub)" <gi...@apache.org> on 2026/05/08 21:36:54 UTC
[PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Klose6 opened a new pull request, #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345
CdcPublisher subscribes to several event-bus addresses (token-range changed, range gained/lost, server stop, CDC cache warmed up) via vertx.eventBus().localConsumer(), so handle(Message) runs on the event loop. The four range/stop branches previously called handleTokenRangeChange / handleRangeGained / handleRangeLost / stop
synchronously, all of which transitively perform blocking work(closing the Kafka producer, stopping SidecarCdc consumers, querying virtual tables). This blocks the event loop and can starve other Vert.x consumers under topology churn.
Dispatch those four branches to the shared worker pool via executorPools.executeBlocking(), matching the existing
ConfigChangedHandler pattern. Keep ON_CDC_CACHE_WARMED_UP on the event loop since it is only a single volatile write. Drop the now-redundant synchronized from handle() (event-loop-dispatched consumers cannot race with themselves; the actual handler methods remain synchronized).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4546841959
Not yet ready to merge. For feature/code changes, the policy requires 2 committers' approval. That is why I requested review from @jyothsnakonisa
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3289642676
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -160,10 +160,25 @@ public void handle(Message<Object> event)
executorPools.executeBlocking(() -> {
restart();
return null;
- });
+ }).onFailure(t -> handleAsyncFailure(ON_CDC_CONFIGURATION_CHANGED.address(), t));
Review Comment:
sure, updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3289647383
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -306,26 +321,35 @@ else if (conf.cdcEnabled())
// EventBus handlers
@Override
- public synchronized void handle(Message<Object> msg)
+ public void handle(Message<Object> msg)
{
- if (msg.address().equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
+ String address = msg.address();
+ if (address.equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
{
- handleTokenRangeChange();
+ executorPools.executeBlocking(() -> { handleTokenRangeChange(); return null; })
Review Comment:
agree, updated, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4546777229
@skoppu22 could you merge this PR since its has 2 approval and also the [Circle CI ](https://app.circleci.com/pipelines/github/Klose6/cassandra-sidecar/12/workflows/ec708044-2522-4b61-9c4e-1a914ca6afff)passed? thanks! cc: @yifan-c
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4567255251
> @Klose6 for the extended message can you give me your name?
>
> patch by "Your name"; reviewed by Jyothsna Konisa, Shailaja Koppu, Yifan Cai for [CASSSIDECAR-452](https://issues.apache.org/jira/browse/CASSSIDECAR-452)
Yafeng Lu, you can find the assignee on that ticket: [CASSSIDECAR-452](https://issues.apache.org/jira/browse/CASSSIDECAR-452), thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3268606797
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -321,26 +321,30 @@ else if (conf.cdcEnabled())
// EventBus handlers
@Override
- public synchronized void handle(Message<Object> msg)
+ public void handle(Message<Object> msg)
{
- if (msg.address().equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
+ String address = msg.address();
+ if (address.equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
{
- handleTokenRangeChange();
+ executorPools.executeBlocking(() -> { handleTokenRangeChange(); return null; });
}
- else if (msg.address().equals(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_GAINED.address()))
+ else if (address.equals(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_GAINED.address()))
{
- handleRangeGained((RangeManager.RangeChangeEvent) msg.body());
+ RangeManager.RangeChangeEvent event = (RangeManager.RangeChangeEvent) msg.body();
+ executorPools.executeBlocking(() -> { handleRangeGained(event); return null; });
}
- else if (msg.address().equals(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_LOST.address()))
+ else if (address.equals(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_LOST.address()))
{
- handleRangeLost((RangeManager.RangeChangeEvent) msg.body());
+ RangeManager.RangeChangeEvent event = (RangeManager.RangeChangeEvent) msg.body();
+ executorPools.executeBlocking(() -> { handleRangeLost(event); return null; });
}
- else if (msg.address().equals(ON_SERVER_STOP.address()))
+ else if (address.equals(ON_SERVER_STOP.address()))
{
- stop();
+ executorPools.executeBlocking(() -> { stop(); return null; });
Review Comment:
Before this change, stop() would have completed before the event loop closed. After this change, there is a possibility of stop() not completing before event loop closed. Better to keep stop() on the event loop, it is not doing heavy work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4511837822
Hi @yifan-c could you also take a look of this PR when you got a chance(since it needs another approval)? thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4567068721
@jyothsnakonisa thank you! I just rebased again and only change in the CHANGES.text.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4550660586
rebase and the new [Circle CI](https://app.circleci.com/pipelines/github/Klose6/cassandra-sidecar/13/workflows/fa3a0558-0bc3-43e0-86bb-7f3c159bddb1)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3289626059
##########
server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java:
##########
@@ -160,4 +175,147 @@ void testEventConsumerCreatesValidConsumer()
assertThat(result).isNotNull();
assertThat(result).isInstanceOf(CdcEventConsumer.class);
}
+
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeChangedDispatchesToWorkerPool() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address());
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleTokenRangeChange();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeGainedPassesEventToHandler() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_GAINED.address());
+ RangeManager.RangeChangeEvent event = mock(RangeManager.RangeChangeEvent.class);
+ when(msg.body()).thenReturn(event);
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleRangeGained(event);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeLostPassesEventToHandler() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_LOST.address());
+ RangeManager.RangeChangeEvent event = mock(RangeManager.RangeChangeEvent.class);
+ when(msg.body()).thenReturn(event);
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleRangeLost(event);
+ }
+
+ @Test
+ void testHandleServerStopRunsStopOnEventLoopWithoutWorkerPool()
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(ON_SERVER_STOP.address());
+
+ // ON_SERVER_STOP must run synchronously on the event loop so shutdown
+ // completes before the loop closes; the no-worker-pool assertion below
+ // is the load-bearing part of the contract.
+ assertThatCode(() -> cdcPublisher.handle(msg)).doesNotThrowAnyException();
+ verify(taskExecutorPool, never()).executeBlocking(any(Callable.class));
Review Comment:
to verify the stop we need to change the stop() from private to package private, I made the change to relax its visibility for testing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "jyothsnakonisa (via GitHub)" <gi...@apache.org>.
jyothsnakonisa commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4566942864
@Klose6 I can help you merge it, can u rebase your PR before that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3269098670
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -321,26 +321,30 @@ else if (conf.cdcEnabled())
// EventBus handlers
@Override
- public synchronized void handle(Message<Object> msg)
+ public void handle(Message<Object> msg)
{
- if (msg.address().equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
+ String address = msg.address();
+ if (address.equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
{
- handleTokenRangeChange();
+ executorPools.executeBlocking(() -> { handleTokenRangeChange(); return null; });
}
- else if (msg.address().equals(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_GAINED.address()))
+ else if (address.equals(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_GAINED.address()))
{
- handleRangeGained((RangeManager.RangeChangeEvent) msg.body());
+ RangeManager.RangeChangeEvent event = (RangeManager.RangeChangeEvent) msg.body();
+ executorPools.executeBlocking(() -> { handleRangeGained(event); return null; });
}
- else if (msg.address().equals(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_LOST.address()))
+ else if (address.equals(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_LOST.address()))
{
- handleRangeLost((RangeManager.RangeChangeEvent) msg.body());
+ RangeManager.RangeChangeEvent event = (RangeManager.RangeChangeEvent) msg.body();
+ executorPools.executeBlocking(() -> { handleRangeLost(event); return null; });
}
- else if (msg.address().equals(ON_SERVER_STOP.address()))
+ else if (address.equals(ON_SERVER_STOP.address()))
{
- stop();
+ executorPools.executeBlocking(() -> { stop(); return null; });
Review Comment:
yeah, it is just update the iterators, no heavy work, let me revert this part.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "jyothsnakonisa (via GitHub)" <gi...@apache.org>.
jyothsnakonisa merged PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4566927119
@Klose6 I don't have merge permission, @yifan-c or @jyothsnakonisa should be able to merge
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3268637091
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -321,26 +321,30 @@ else if (conf.cdcEnabled())
// EventBus handlers
@Override
- public synchronized void handle(Message<Object> msg)
+ public void handle(Message<Object> msg)
{
- if (msg.address().equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
+ String address = msg.address();
+ if (address.equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
{
- handleTokenRangeChange();
+ executorPools.executeBlocking(() -> { handleTokenRangeChange(); return null; });
Review Comment:
We need to check for onFailure for all these executeBlocking calls, to catch any exceptions and handle them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3269092955
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -321,26 +321,30 @@ else if (conf.cdcEnabled())
// EventBus handlers
@Override
- public synchronized void handle(Message<Object> msg)
+ public void handle(Message<Object> msg)
{
- if (msg.address().equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
+ String address = msg.address();
+ if (address.equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
{
- handleTokenRangeChange();
+ executorPools.executeBlocking(() -> { handleTokenRangeChange(); return null; });
Review Comment:
agree, let me add the logic to catch the error and print the log.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3283610939
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
Review Comment:
sure, just added, pls review again! thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3283153204
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
Review Comment:
[minor] I understand this is not your change, but would be great if we can add onFailure for this as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4514760926
Circle CI: https://app.circleci.com/pipelines/github/Klose6/cassandra-sidecar/10/workflows/f28fd6f8-1857-413e-8ec1-070bd431d59d
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3286554272
##########
server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java:
##########
@@ -160,4 +175,147 @@ void testEventConsumerCreatesValidConsumer()
assertThat(result).isNotNull();
assertThat(result).isInstanceOf(CdcEventConsumer.class);
}
+
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeChangedDispatchesToWorkerPool() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address());
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleTokenRangeChange();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeGainedPassesEventToHandler() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_GAINED.address());
+ RangeManager.RangeChangeEvent event = mock(RangeManager.RangeChangeEvent.class);
+ when(msg.body()).thenReturn(event);
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleRangeGained(event);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeLostPassesEventToHandler() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_LOST.address());
+ RangeManager.RangeChangeEvent event = mock(RangeManager.RangeChangeEvent.class);
+ when(msg.body()).thenReturn(event);
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleRangeLost(event);
+ }
+
+ @Test
+ void testHandleServerStopRunsStopOnEventLoopWithoutWorkerPool()
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(ON_SERVER_STOP.address());
+
+ // ON_SERVER_STOP must run synchronously on the event loop so shutdown
+ // completes before the loop closes; the no-worker-pool assertion below
+ // is the load-bearing part of the contract.
+ assertThatCode(() -> cdcPublisher.handle(msg)).doesNotThrowAnyException();
+ verify(taskExecutorPool, never()).executeBlocking(any(Callable.class));
+ }
+
+ @Test
+ void testHandleCdcCacheWarmedUpRunsOnEventLoopWithoutWorkerPool()
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(ON_CDC_CACHE_WARMED_UP.address());
+
+ cdcPublisher.handle(msg);
+
+ verify(taskExecutorPool, never()).executeBlocking(any(Callable.class));
+ }
+
+ @Test
+ void testHandleUnknownAddressIsNoOp()
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn("unknown.address");
+
+ cdcPublisher.handle(msg);
+
+ verify(taskExecutorPool, never()).executeBlocking(any(Callable.class));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleWorkerPoolFailureIsLoggedAndCaptured()
+ {
+ RuntimeException boom = new RuntimeException("boom");
+ when(taskExecutorPool.executeBlocking(any(Callable.class))).thenReturn(Future.failedFuture(boom));
+
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address());
+
+ cdcPublisher.handle(msg);
+
+ // The discarded Future's failure path must reach our backstop:
+ // captureUnrecoverableCdcError so the failure is observable in metrics.
+ verify(sidecarCdcStats).captureUnrecoverableCdcError(boom);
+ }
+
+ private SslConfiguration mockSslConfiguration(boolean enabled,
+ boolean preferOpenSSL,
+ String clientAuth,
+ java.util.List<String> cipherSuites,
+ java.util.List<String> secureTransportProtocols,
+ String handshakeTimeout,
+ boolean keystoreConfigured,
+ boolean truststoreConfigured)
+ {
+ SslConfiguration sslConfig = mock(SslConfiguration.class, RETURNS_DEEP_STUBS);
+ when(sslConfig.enabled()).thenReturn(enabled);
+ when(sslConfig.preferOpenSSL()).thenReturn(preferOpenSSL);
+ when(sslConfig.clientAuth()).thenReturn(clientAuth);
+ when(sslConfig.cipherSuites()).thenReturn(cipherSuites);
+ when(sslConfig.secureTransportProtocols()).thenReturn(secureTransportProtocols);
+
+ SecondBoundConfiguration durationSpec = mock(SecondBoundConfiguration.class);
+ when(durationSpec.toString()).thenReturn(handshakeTimeout);
+ when(sslConfig.handshakeTimeout()).thenReturn(durationSpec);
+
+ when(sslConfig.isKeystoreConfigured()).thenReturn(keystoreConfigured);
+ when(sslConfig.isTrustStoreConfigured()).thenReturn(truststoreConfigured);
+
+ return sslConfig;
+ }
+
+ private KeyStoreConfiguration mockKeystoreConfiguration(String path, String password, String type)
+ {
+ KeyStoreConfiguration config = mock(KeyStoreConfiguration.class);
+ when(config.path()).thenReturn(path);
+ when(config.password()).thenReturn(password);
+ when(config.type()).thenReturn(type);
+ return config;
+ }
Review Comment:
those 2 methods are unused. Please remove them if it is a mistake.
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -306,26 +321,35 @@ else if (conf.cdcEnabled())
// EventBus handlers
@Override
- public synchronized void handle(Message<Object> msg)
+ public void handle(Message<Object> msg)
{
- if (msg.address().equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
+ String address = msg.address();
+ if (address.equals(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address()))
{
- handleTokenRangeChange();
+ executorPools.executeBlocking(() -> { handleTokenRangeChange(); return null; })
Review Comment:
Code looks nicer with `runBlocking`. Please update the other places in the method.
```suggestion
executorPools.runBlocking(this::handleTokenRangeChange)
```
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -160,10 +160,25 @@ public void handle(Message<Object> event)
executorPools.executeBlocking(() -> {
restart();
return null;
- });
+ }).onFailure(t -> handleAsyncFailure(ON_CDC_CONFIGURATION_CHANGED.address(), t));
}
}
+ /**
+ * Backstop for fire-and-forget {@code executeBlocking(...)} dispatches in event-bus handlers.
+ * The handler methods ({@code restart}, {@code stop}) already catch and record their own
+ * failures via {@code sidecarCdcStats}; this fires only when something unexpected escapes
+ * (e.g. an {@link Error}, or a stats-counter increment that throws before reaching the
+ * handler's own try/catch). Logging at ERROR + bumping
+ * {@link SidecarCdcStats#captureUnrecoverableCdcError(Throwable)} so the issue is surfaced
+ * rather than silently dropped by the discarded {@code Future}.
+ */
+ private void handleAsyncFailure(String address, Throwable t)
+ {
+ LOGGER.error("Unexpected error while processing event '{}' on worker pool", address, t);
+ sidecarCdcStats.captureUnrecoverableCdcError(t);
+ }
+
Review Comment:
I do not think the method will be invoked, since the `restart` & `stop` methods does not re-throw exception. I am fine with keeping the method.
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -160,10 +160,25 @@ public void handle(Message<Object> event)
executorPools.executeBlocking(() -> {
restart();
return null;
- });
+ }).onFailure(t -> handleAsyncFailure(ON_CDC_CONFIGURATION_CHANGED.address(), t));
Review Comment:
cosmetic feedback
```suggestion
executorPools.runBlocking(CdcPublisher.this::restart)
.onFailure(t -> handleAsyncFailure(ON_CDC_CONFIGURATION_CHANGED.address(), t));
```
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -160,10 +160,25 @@ public void handle(Message<Object> event)
executorPools.executeBlocking(() -> {
restart();
return null;
- });
+ }).onFailure(t -> handleAsyncFailure(ON_CDC_CONFIGURATION_CHANGED.address(), t));
}
}
+ /**
+ * Backstop for fire-and-forget {@code executeBlocking(...)} dispatches in event-bus handlers.
+ * The handler methods ({@code restart}, {@code stop}) already catch and record their own
+ * failures via {@code sidecarCdcStats}; this fires only when something unexpected escapes
+ * (e.g. an {@link Error}, or a stats-counter increment that throws before reaching the
+ * handler's own try/catch). Logging at ERROR + bumping
+ * {@link SidecarCdcStats#captureUnrecoverableCdcError(Throwable)} so the issue is surfaced
+ * rather than silently dropped by the discarded {@code Future}.
+ */
Review Comment:
nit: the method itself is pretty straightforward. I find the comment is harder to understand.
##########
server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java:
##########
@@ -160,4 +175,147 @@ void testEventConsumerCreatesValidConsumer()
assertThat(result).isNotNull();
assertThat(result).isInstanceOf(CdcEventConsumer.class);
}
+
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeChangedDispatchesToWorkerPool() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address());
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleTokenRangeChange();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeGainedPassesEventToHandler() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_GAINED.address());
+ RangeManager.RangeChangeEvent event = mock(RangeManager.RangeChangeEvent.class);
+ when(msg.body()).thenReturn(event);
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleRangeGained(event);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeLostPassesEventToHandler() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_LOST.address());
+ RangeManager.RangeChangeEvent event = mock(RangeManager.RangeChangeEvent.class);
+ when(msg.body()).thenReturn(event);
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleRangeLost(event);
+ }
+
+ @Test
+ void testHandleServerStopRunsStopOnEventLoopWithoutWorkerPool()
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(ON_SERVER_STOP.address());
+
+ // ON_SERVER_STOP must run synchronously on the event loop so shutdown
+ // completes before the loop closes; the no-worker-pool assertion below
+ // is the load-bearing part of the contract.
+ assertThatCode(() -> cdcPublisher.handle(msg)).doesNotThrowAnyException();
+ verify(taskExecutorPool, never()).executeBlocking(any(Callable.class));
Review Comment:
Spy the publisher to verify stop is called too.
```suggestion
CdcPublisher spyPublisher = spy(cdcPublisher);
spyPublisher.handle(msg);
verify(spyPublisher).stop(); // verify stop is called
verify(taskExecutorPool, never()).executeBlocking(any(Callable.class));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3289649761
##########
server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java:
##########
@@ -160,4 +175,147 @@ void testEventConsumerCreatesValidConsumer()
assertThat(result).isNotNull();
assertThat(result).isInstanceOf(CdcEventConsumer.class);
}
+
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeChangedDispatchesToWorkerPool() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address());
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleTokenRangeChange();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeGainedPassesEventToHandler() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_GAINED.address());
+ RangeManager.RangeChangeEvent event = mock(RangeManager.RangeChangeEvent.class);
+ when(msg.body()).thenReturn(event);
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleRangeGained(event);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleTokenRangeLostPassesEventToHandler() throws Exception
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_LOST.address());
+ RangeManager.RangeChangeEvent event = mock(RangeManager.RangeChangeEvent.class);
+ when(msg.body()).thenReturn(event);
+
+ CdcPublisher spyPublisher = spy(cdcPublisher);
+ spyPublisher.handle(msg);
+
+ ArgumentCaptor<Callable<Void>> captor = ArgumentCaptor.forClass(Callable.class);
+ verify(taskExecutorPool).executeBlocking(captor.capture());
+
+ captor.getValue().call();
+ verify(spyPublisher).handleRangeLost(event);
+ }
+
+ @Test
+ void testHandleServerStopRunsStopOnEventLoopWithoutWorkerPool()
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(ON_SERVER_STOP.address());
+
+ // ON_SERVER_STOP must run synchronously on the event loop so shutdown
+ // completes before the loop closes; the no-worker-pool assertion below
+ // is the load-bearing part of the contract.
+ assertThatCode(() -> cdcPublisher.handle(msg)).doesNotThrowAnyException();
+ verify(taskExecutorPool, never()).executeBlocking(any(Callable.class));
+ }
+
+ @Test
+ void testHandleCdcCacheWarmedUpRunsOnEventLoopWithoutWorkerPool()
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(ON_CDC_CACHE_WARMED_UP.address());
+
+ cdcPublisher.handle(msg);
+
+ verify(taskExecutorPool, never()).executeBlocking(any(Callable.class));
+ }
+
+ @Test
+ void testHandleUnknownAddressIsNoOp()
+ {
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn("unknown.address");
+
+ cdcPublisher.handle(msg);
+
+ verify(taskExecutorPool, never()).executeBlocking(any(Callable.class));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testHandleWorkerPoolFailureIsLoggedAndCaptured()
+ {
+ RuntimeException boom = new RuntimeException("boom");
+ when(taskExecutorPool.executeBlocking(any(Callable.class))).thenReturn(Future.failedFuture(boom));
+
+ Message<Object> msg = mock(Message.class);
+ when(msg.address()).thenReturn(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address());
+
+ cdcPublisher.handle(msg);
+
+ // The discarded Future's failure path must reach our backstop:
+ // captureUnrecoverableCdcError so the failure is observable in metrics.
+ verify(sidecarCdcStats).captureUnrecoverableCdcError(boom);
+ }
+
+ private SslConfiguration mockSslConfiguration(boolean enabled,
+ boolean preferOpenSSL,
+ String clientAuth,
+ java.util.List<String> cipherSuites,
+ java.util.List<String> secureTransportProtocols,
+ String handshakeTimeout,
+ boolean keystoreConfigured,
+ boolean truststoreConfigured)
+ {
+ SslConfiguration sslConfig = mock(SslConfiguration.class, RETURNS_DEEP_STUBS);
+ when(sslConfig.enabled()).thenReturn(enabled);
+ when(sslConfig.preferOpenSSL()).thenReturn(preferOpenSSL);
+ when(sslConfig.clientAuth()).thenReturn(clientAuth);
+ when(sslConfig.cipherSuites()).thenReturn(cipherSuites);
+ when(sslConfig.secureTransportProtocols()).thenReturn(secureTransportProtocols);
+
+ SecondBoundConfiguration durationSpec = mock(SecondBoundConfiguration.class);
+ when(durationSpec.toString()).thenReturn(handshakeTimeout);
+ when(sslConfig.handshakeTimeout()).thenReturn(durationSpec);
+
+ when(sslConfig.isKeystoreConfigured()).thenReturn(keystoreConfigured);
+ when(sslConfig.isTrustStoreConfigured()).thenReturn(truststoreConfigured);
+
+ return sslConfig;
+ }
+
+ private KeyStoreConfiguration mockKeystoreConfiguration(String path, String password, String type)
+ {
+ KeyStoreConfiguration config = mock(KeyStoreConfiguration.class);
+ when(config.path()).thenReturn(path);
+ when(config.password()).thenReturn(password);
+ when(config.type()).thenReturn(type);
+ return config;
+ }
Review Comment:
yeah, good catch, removed them now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "Klose6 (via GitHub)" <gi...@apache.org>.
Klose6 commented on code in PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#discussion_r3289653401
##########
server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java:
##########
@@ -160,10 +160,25 @@ public void handle(Message<Object> event)
executorPools.executeBlocking(() -> {
restart();
return null;
- });
+ }).onFailure(t -> handleAsyncFailure(ON_CDC_CONFIGURATION_CHANGED.address(), t));
}
}
+ /**
+ * Backstop for fire-and-forget {@code executeBlocking(...)} dispatches in event-bus handlers.
+ * The handler methods ({@code restart}, {@code stop}) already catch and record their own
+ * failures via {@code sidecarCdcStats}; this fires only when something unexpected escapes
+ * (e.g. an {@link Error}, or a stats-counter increment that throws before reaching the
+ * handler's own try/catch). Logging at ERROR + bumping
+ * {@link SidecarCdcStats#captureUnrecoverableCdcError(Throwable)} so the issue is surfaced
+ * rather than silently dropped by the discarded {@code Future}.
+ */
Review Comment:
change it to simple words.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-452: Avoid blocking the event loop in CdcPublisher event-bus handler [cassandra-sidecar]
Posted by "jyothsnakonisa (via GitHub)" <gi...@apache.org>.
jyothsnakonisa commented on PR #345:
URL: https://github.com/apache/cassandra-sidecar/pull/345#issuecomment-4567106671
@Klose6 for the extended message can you give me your name?
patch by <Your name>; reviewed by Jyothsna Konisa, Shailaja Koppu, Yifan Cai for CASSSIDECAR-452
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org