Skip to content

Commit c4b17ec

Browse files
stheppistheppi
authored
Merge pull request #402 from lensesio-dev/fix/http-sink-reporter-shutdown-order
fix(http-sink): close reporting controllers only after consumer fibers stop
2 parents 80443ce + b365f67 commit c4b17ec

3 files changed

Lines changed: 174 additions & 15 deletions

File tree

‎kafka-connect-http/src/main/scala/io/lenses/streamreactor/connect/http/sink/HttpSinkTask.scala‎

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -198,14 +198,11 @@ class HttpSinkTask extends SinkTask with LazyLogging with JarManifestProvided {
198198
(for {
199199
taskNumber <- taskNumberRef.get
200200
_ <- IO(MetricsRegistrar.unregisterMetricsMBean(sinkName, taskNumber))
201-
// Signal termination first, then wait for the consumer fibers to finish (their in-flight
202-
// requests cancelled) before releasing the shared HTTP client, so teardown does not race with
203-
// an in-progress send.
201+
// Signal termination, then let the manager own the teardown order (consumer fibers first,
202+
// then reporting controllers, then the shared HTTP client) so it does not race with an
203+
// in-progress send.
204204
_ <- deferred.complete(().asRight)
205-
_ <- maybeWriterManager.traverse { x =>
206-
x.closeReportingControllers()
207-
x.awaitConsumers *> x.close
208-
}
205+
_ <- maybeWriterManager.traverse_(_.shutdown)
209206
} yield ()).unsafeRunSync()
210207

211208
}

‎kafka-connect-http/src/main/scala/io/lenses/streamreactor/connect/http/sink/HttpWriterManager.scala‎

Lines changed: 17 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -191,7 +191,7 @@ class HttpWriterManager(
191191
template: TemplateType,
192192
httpRequestSender: HttpRequestSender,
193193
batchPolicy: BatchPolicy,
194-
val close: IO[Unit],
194+
private val close: IO[Unit],
195195
writersRef: Ref[IO, Map[Topic, HttpWriter]],
196196
writerCreationLock: Semaphore[IO],
197197
deferred: Deferred[IO, Either[Throwable, Unit]],
@@ -259,7 +259,7 @@ class HttpWriterManager(
259259
* so their in-flight requests are cancelled before the shared HTTP client is released. Bounded by
260260
* a timeout so a stuck request cannot make shutdown hang indefinitely.
261261
*/
262-
def awaitConsumers: IO[Unit] =
262+
private def awaitConsumers: IO[Unit] =
263263
consumerFibersRef.get
264264
.flatMap(fibers => fibers.traverse_(_.join.void))
265265
.timeoutTo(
@@ -268,12 +268,22 @@ class HttpWriterManager(
268268
)
269269

270270
/**
271-
* Closes the reporting controllers.
271+
* Closes the reporting controllers. Must only run once the consumer fibers have stopped:
272+
* `HttpWriter.reportResult` enqueues onto these controllers from an in-flight send, and those
273+
* reports are dropped (and the offer can block on a queue nothing drains) if they are closed first.
272274
*/
273-
def closeReportingControllers(): Unit = {
274-
errorReportingController.close()
275-
successReportingController.close()
276-
}
275+
private def closeReportingControllers: IO[Unit] =
276+
IO(errorReportingController.close()).guarantee(IO(successReportingController.close()))
277+
278+
/**
279+
* Orderly teardown, to be run after the termination signal has been completed. Each stage is a
280+
* finalizer of the previous one, so a failure or timeout part-way still releases everything:
281+
* wait for the consumer fibers (their in-flight requests cancelled), only then stop the
282+
* reporting controllers (so no in-flight send can enqueue into a closed reporter), and finally
283+
* release the shared HTTP client.
284+
*/
285+
def shutdown: IO[Unit] =
286+
awaitConsumers.guarantee(closeReportingControllers).guarantee(close)
277287

278288
/**
279289
* Gets or creates an HTTP writer for the given topic.

‎kafka-connect-http/src/test/scala/io/lenses/streamreactor/connect/http/sink/HttpWriterManagerTest.scala‎

Lines changed: 153 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,14 +15,44 @@
1515
*/
1616
package io.lenses.streamreactor.connect.http.sink;
1717

18+
import cats.data.NonEmptySeq
19+
import cats.effect.Deferred
20+
import cats.effect.IO
21+
import cats.effect.Ref
22+
import cats.effect.std.Semaphore
23+
import cats.effect.unsafe.implicits.global
24+
import cats.implicits.catsSyntaxEitherId
25+
import com.typesafe.scalalogging.LazyLogging
26+
import io.lenses.streamreactor.common.batch.BatchPolicy
27+
import io.lenses.streamreactor.common.batch.Count
28+
import io.lenses.streamreactor.connect.cloud.common.model.Topic
29+
import io.lenses.streamreactor.connect.http.sink.client.HttpRequestSender
30+
import io.lenses.streamreactor.connect.http.sink.client.HttpResponseFailure
31+
import io.lenses.streamreactor.connect.http.sink.client.HttpResponseSuccess
32+
import io.lenses.streamreactor.connect.http.sink.config.ErrorNullPayloadHandler
33+
import io.lenses.streamreactor.connect.http.sink.reporter.model.HttpFailureConnectorSpecificRecordData
34+
import io.lenses.streamreactor.connect.http.sink.reporter.model.HttpSuccessConnectorSpecificRecordData
35+
import io.lenses.streamreactor.connect.http.sink.tpl.Headers
36+
import io.lenses.streamreactor.connect.http.sink.tpl.ProcessedTemplate
37+
import io.lenses.streamreactor.connect.http.sink.tpl.RenderedRecord
38+
import io.lenses.streamreactor.connect.http.sink.tpl.SimpleTemplate
39+
import io.lenses.streamreactor.connect.http.sink.tpl.TemplateType
40+
import io.lenses.streamreactor.connect.reporting.ReportingController
1841
import org.http4s.Response
1942
import org.http4s.Status
2043
import org.http4s.WaitQueueTimeoutException
44+
import org.mockito.ArgumentMatchers.any
45+
import org.mockito.MockitoSugar
46+
import org.mockito.invocation.InvocationOnMock
2147
import org.scalatest.EitherValues
2248
import org.scalatest.funsuite.AnyFunSuiteLike
2349
import org.scalatest.matchers.should.Matchers
2450

25-
class HttpWriterManagerTest extends AnyFunSuiteLike with Matchers with EitherValues {
51+
import java.util.concurrent.ConcurrentLinkedQueue
52+
import scala.concurrent.duration.DurationInt
53+
import scala.jdk.CollectionConverters._
54+
55+
class HttpWriterManagerTest extends AnyFunSuiteLike with Matchers with EitherValues with MockitoSugar with LazyLogging {
2656

2757
test("isErrorOrStatus returns true if the statusCodes is matched") {
2858
val statusCodes = Set(408, 429)
@@ -46,4 +76,126 @@ class HttpWriterManagerTest extends AnyFunSuiteLike with Matchers with EitherVal
4676

4777
HttpWriterManager.isErrorOrRetriableStatus(Left(WaitQueueTimeoutException), statusCodes) should be(false)
4878
}
79+
80+
// Covers the PR-review finding: `HttpSinkTask.stop()` used to close the reporting controllers
81+
// before waiting for the consumer fibers, so an in-flight `sendBatch` could still reach
82+
// `reportResult` and enqueue onto a controller that was already shutting down. `shutdown` now owns
83+
// the whole teardown order, so these tests exercise it directly instead of `HttpSinkTask`.
84+
85+
private val sinkName = "MySinkName"
86+
private val topic = Topic("myTopic")
87+
private val record = RenderedRecord(topic.withPartition(1).atOffset(1L), 1L, "record", Seq.empty, "")
88+
89+
private def successTemplate(): TemplateType =
90+
SimpleTemplate("http://bench.invalid",
91+
"content",
92+
Headers(Seq.empty, copyMessageHeaders = false),
93+
ErrorNullPayloadHandler,
94+
)
95+
96+
private def buildManager(
97+
sender: HttpRequestSender,
98+
errorController: ReportingController[HttpFailureConnectorSpecificRecordData],
99+
successController: ReportingController[HttpSuccessConnectorSpecificRecordData],
100+
closeIO: IO[Unit],
101+
deferred: Deferred[IO, Either[Throwable, Unit]],
102+
): IO[HttpWriterManager] =
103+
for {
104+
writersRef <- Ref.of[IO, Map[Topic, HttpWriter]](Map.empty)
105+
lock <- Semaphore[IO](1)
106+
} yield new HttpWriterManager(
107+
sinkName = sinkName,
108+
template = successTemplate(),
109+
httpRequestSender = sender,
110+
batchPolicy = BatchPolicy(logger, Count(1)),
111+
close = closeIO,
112+
writersRef = writersRef,
113+
writerCreationLock = lock,
114+
deferred = deferred,
115+
errorThreshold = 5,
116+
uploadSyncPeriod = 0,
117+
tidyJson = false,
118+
errorReportingController = errorController,
119+
successReportingController = successController,
120+
maxQueueSize = 10,
121+
maxQueueOfferTimeout = 1.minute,
122+
)
123+
124+
test("shutdown does not close the reporting controllers until it is actually run") {
125+
val errorController = mock[ReportingController[HttpFailureConnectorSpecificRecordData]]
126+
val successController = mock[ReportingController[HttpSuccessConnectorSpecificRecordData]]
127+
128+
val io = for {
129+
deferred <- Deferred[IO, Either[Throwable, Unit]]
130+
manager <- buildManager(mock[HttpRequestSender], errorController, successController, IO.unit, deferred)
131+
_ <- IO(verify(errorController, never).close())
132+
_ <- IO(verify(successController, never).close())
133+
_ <- deferred.complete(().asRight)
134+
_ <- manager.shutdown
135+
} yield {
136+
verify(errorController).close()
137+
verify(successController).close()
138+
}
139+
140+
io.unsafeRunSync()
141+
}
142+
143+
test("shutdown closes the success controller and releases the client even when the error controller's close throws") {
144+
val errorController = mock[ReportingController[HttpFailureConnectorSpecificRecordData]]
145+
when(errorController.close()).thenThrow(new RuntimeException("boom"))
146+
val successController = mock[ReportingController[HttpSuccessConnectorSpecificRecordData]]
147+
val closed = new ConcurrentLinkedQueue[String]()
148+
149+
val io = for {
150+
deferred <- Deferred[IO, Either[Throwable, Unit]]
151+
manager <- buildManager(mock[HttpRequestSender],
152+
errorController,
153+
successController,
154+
IO(closed.add("close")).void,
155+
deferred,
156+
)
157+
_ <- deferred.complete(().asRight)
158+
result <- manager.shutdown.attempt
159+
} yield {
160+
result.left.value.getMessage shouldBe "boom"
161+
verify(successController).close()
162+
closed.asScala.toList shouldBe List("close")
163+
}
164+
165+
io.unsafeRunSync()
166+
}
167+
168+
test(
169+
"shutdown waits for the consumer fiber's in-flight send to be cancelled before closing the reporters and the client",
170+
) {
171+
val log = new ConcurrentLinkedQueue[String]()
172+
val errorController = mock[ReportingController[HttpFailureConnectorSpecificRecordData]]
173+
when(errorController.close()).thenAnswer { (_: InvocationOnMock) => log.add("errorClose"); () }
174+
val successController = mock[ReportingController[HttpSuccessConnectorSpecificRecordData]]
175+
when(successController.close()).thenAnswer { (_: InvocationOnMock) => log.add("successClose"); () }
176+
val sender = mock[HttpRequestSender]
177+
178+
val io = for {
179+
entered <- Deferred[IO, Unit]
180+
deferred <- Deferred[IO, Either[Throwable, Unit]]
181+
// The send never completes on its own: it is only ever unblocked by the cancellation that
182+
// `HttpWriterManager.startConsumer`'s `IO.race(consumer, deferred.get)` delivers once
183+
// `deferred` is completed, mirroring the real shutdown path exactly.
184+
_ <- IO {
185+
when(sender.sendHttpRequest(any[ProcessedTemplate])).thenReturn(
186+
entered.complete(()) *> IO.never[Either[HttpResponseFailure, HttpResponseSuccess]]
187+
.onCancel(IO(log.add("consumerCancelled")).void),
188+
)
189+
}
190+
manager <- buildManager(sender, errorController, successController, IO(log.add("close")).void, deferred)
191+
writer <- manager.getWriter(topic)
192+
_ <- writer.add(NonEmptySeq.of(record))
193+
// The batch is now stuck mid-send, holding its permit.
194+
_ <- entered.get
195+
_ <- deferred.complete(().asRight)
196+
_ <- manager.shutdown
197+
} yield log.asScala.toList shouldBe List("consumerCancelled", "errorClose", "successClose", "close")
198+
199+
io.unsafeRunSync()
200+
}
49201
}

0 commit comments

Comments
 (0)