lhotari commented on code in PR #23853:
URL: https://github.com/apache/pulsar/pull/23853#discussion_r1918243682
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java:
##########
@@ -192,13 +193,12 @@ public void connectionClosed(ClientCnx cnx,
Optional<Long> initialConnectionDela
duringConnect.set(false);
state.client.getCnxPool().releaseConnection(cnx);
if (CLIENT_CNX_UPDATER.compareAndSet(this, cnx, null)) {
- if (!isValidStateForReconnection()) {
+ if (!state.changeToConnecting()) {
Review Comment:
previously this would allow the connection to be in `Ready` state while
switching to `Connecting`. Is it intentional that it's no longer allowed?
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java:
##########
@@ -74,35 +66,4 @@ public void testPopulateMessageSchema() {
assertTrue(producer.populateMessageSchema(msg, null));
verify(msg).setSchemaState(MessageImpl.SchemaState.Ready);
}
-
- @Test
- public void testClearPendingMessageWhenCloseAsync() {
- PulsarClientImpl client = mock(PulsarClientImpl.class);
- Mockito.doReturn(1L).when(client).newProducerId();
- ClientConfigurationData clientConf = new ClientConfigurationData();
- clientConf.setStatsIntervalSeconds(-1);
- Mockito.doReturn(clientConf).when(client).getConfiguration();
- Mockito.doReturn(new
InstrumentProvider(null)).when(client).instrumentProvider();
- ConnectionPool connectionPool = mock(ConnectionPool.class);
- Mockito.doReturn(1).when(connectionPool).genRandomKeyToSelectCon();
- Mockito.doReturn(connectionPool).when(client).getCnxPool();
- HashedWheelTimer timer = mock(HashedWheelTimer.class);
- Mockito.doReturn(null).when(timer).newTimeout(Mockito.any(),
Mockito.anyLong(), Mockito.any());
- Mockito.doReturn(timer).when(client).timer();
- ProducerConfigurationData producerConf = new
ProducerConfigurationData();
- producerConf.setSendTimeoutMs(-1);
- ProducerImpl<?> producer = Mockito.spy(new ProducerImpl<>(client,
"topicName", producerConf, null, 0, null, null, Optional.empty()));
-
- // make sure throw exception when send request to broker
- ClientCnx clientCnx = mock(ClientCnx.class);
- CompletableFuture<ProducerResponse> tCompletableFuture = new
CompletableFuture<>();
- tCompletableFuture.completeExceptionally(new
PulsarClientException("error"));
- when(clientCnx.sendRequestWithId(Mockito.any(),
Mockito.anyLong())).thenReturn(tCompletableFuture);
- Mockito.doReturn(clientCnx).when(producer).cnx();
-
- // run closeAsync and verify
- CompletableFuture<Void> voidCompletableFuture = producer.closeAsync();
- verify(producer).closeAndClearPendingMessages();
- }
Review Comment:
Why is the test case completely removed? Would it be possible to ensure with
a test that the problem described in PR #23761 is addressed?
--
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: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
