You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Gerrrr (via GitHub)" <gi...@apache.org> on 2023/09/05 23:31:53 UTC
[GitHub] [flink-connector-kafka] Gerrrr opened a new pull request, #46: [FLINK-33017] Remove dependency on shaded guava
Gerrrr opened a new pull request, #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46
The bump in shaded guava in Flink 1.18 changed import paths and caused the class loader fail when loading ManagedMemoryUtils.
Looking at the root cause of the issue, shading was used as a technique to avoid dependency hell. As flink-connector-kafka should work with both flink 1.17 and 1.18 that use different guava versions (and hence library import paths), shading did not really solve the problem it was introduced for in the first place.
There are several several options to work around the problem. First, we could introduce our own shading for guava. Second, we could see if the dependency on guava is necessary at all and maybe remove it completely.
This patch takes the latter route and removes dependency on guava from this connector.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai closed pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai closed pull request #46: [FLINK-33017] Remove dependency on shaded guava
URL: https://github.com/apache/flink-connector-kafka/pull/46
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317629973
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -104,9 +103,10 @@ class KafkaWriter<IN>
// producer pool only used for exactly once
private final Deque<FlinkKafkaInternalProducer<byte[], byte[]>> producerPool =
new ArrayDeque<>();
- private final Closer closer = Closer.create();
private long lastCheckpointId;
+ private final Set<AutoCloseable> closeables = new HashSet<>();
Review Comment:
Lets rename this to maybe `producerCloseables` to be more exact. Will make understanding the `close()` method a bit easier.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317636517
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -104,9 +103,10 @@ class KafkaWriter<IN>
// producer pool only used for exactly once
private final Deque<FlinkKafkaInternalProducer<byte[], byte[]>> producerPool =
new ArrayDeque<>();
- private final Closer closer = Closer.create();
private long lastCheckpointId;
+ private final Set<AutoCloseable> closeables = new HashSet<>();
Review Comment:
Also, should this be a `Queue` / `List` instead? Here we never want actual implementation of the producer object's equality check to interfere here and leave room to potentially have a producer left unclosed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] Gerrrr commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "Gerrrr (via GitHub)" <gi...@apache.org>.
Gerrrr commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1316546581
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/shuffle/KafkaShuffleITCase.java:
##########
@@ -374,17 +371,16 @@ private void testRecordSerDe(TimeCharacteristic timeCharacteristic) throws Excep
// Records in a single partition are kept in order
Collection<ConsumerRecord<byte[], byte[]>> records =
- Iterables.getOnlyElement(
- testKafkaShuffleProducer(
- topic(
- "test_serde-" + UUID.randomUUID(),
- timeCharacteristic),
- env,
- 1,
- 1,
- numElementsPerProducer,
- timeCharacteristic)
- .values());
+ testKafkaShuffleProducer(
+ topic("test_serde-" + UUID.randomUUID(), timeCharacteristic),
+ env,
+ 1,
+ 1,
+ numElementsPerProducer,
+ timeCharacteristic)
+ .values()
+ .iterator()
+ .next();
Review Comment:
Done in 2b184daf3ad0700628d8c90edd24db4349db5285.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] Gerrrr commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "Gerrrr (via GitHub)" <gi...@apache.org>.
Gerrrr commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317654030
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -245,12 +245,12 @@ public void close() throws Exception {
LOG.debug("Closing writer with {}", currentProducer);
closeAll(
this::abortCurrentProducer,
- closer,
producerPool::clear,
() -> {
checkState(currentProducer.isClosed());
currentProducer = null;
});
+ closeAll(closeables);
Review Comment:
Fixed in [dfcfdb4](https://github.com/apache/flink-connector-kafka/pull/46/commits/dfcfdb47fabf63632edac94139f9acb3c36eb3b5). I also added a meaningful error message in [96bd344](https://github.com/apache/flink-connector-kafka/pull/46/commits/96bd3447646d777eba1a4bb808e6e16c8f74ee77) to make `java.lang.IllegalStateException: null` a tad user-friendlier.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1316526004
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -452,12 +453,9 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
asyncProducerException = decorateException(metadata, exception, producer);
}
+ // Checking for exceptions from previous writes
mailboxExecutor.submit(
- () -> {
- // Checking for exceptions from previous writes
- checkAsyncException();
- },
- "Update error metric");
+ KafkaWriter.this::checkAsyncException, "Update error metric");
Review Comment:
minor nit:
was this necessary? Perhaps could have been a separate `[hotfix]` commit.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java:
##########
@@ -173,9 +172,10 @@ private static Properties getProperties() {
}
private static List<Consumer<FlinkKafkaInternalProducer<?, ?>>> provideTransactionsFinalizer() {
- return Lists.newArrayList(
- FlinkKafkaInternalProducer::commitTransaction,
- FlinkKafkaInternalProducer::abortTransaction);
+ List<Consumer<FlinkKafkaInternalProducer<?, ?>>> ret = new ArrayList<>();
+ ret.add(FlinkKafkaInternalProducer::commitTransaction);
+ ret.add(FlinkKafkaInternalProducer::abortTransaction);
Review Comment:
nit:
a lot of these can probably be replaced with:
```
Arrays.asList(
item1,
item2
)
```
to remove the Java boilerplates.
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/shuffle/KafkaShuffleITCase.java:
##########
@@ -374,17 +371,16 @@ private void testRecordSerDe(TimeCharacteristic timeCharacteristic) throws Excep
// Records in a single partition are kept in order
Collection<ConsumerRecord<byte[], byte[]>> records =
- Iterables.getOnlyElement(
- testKafkaShuffleProducer(
- topic(
- "test_serde-" + UUID.randomUUID(),
- timeCharacteristic),
- env,
- 1,
- 1,
- numElementsPerProducer,
- timeCharacteristic)
- .values());
+ testKafkaShuffleProducer(
+ topic("test_serde-" + UUID.randomUUID(), timeCharacteristic),
+ env,
+ 1,
+ 1,
+ numElementsPerProducer,
+ timeCharacteristic)
+ .values()
+ .iterator()
+ .next();
Review Comment:
(same for some other places where `Iterables.getOnlyElement` was replaced)
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -245,12 +245,12 @@ public void close() throws Exception {
LOG.debug("Closing writer with {}", currentProducer);
closeAll(
this::abortCurrentProducer,
- closer,
producerPool::clear,
() -> {
checkState(currentProducer.isClosed());
currentProducer = null;
});
+ closeAll(closeables);
Review Comment:
originally, with a single `closeAll` we had an implicit ordering on the sequence of what stuff to close after the other.
Now with the two `closeAll`s, the ordering has changed.
Could you double check if moving the closeables to the end would affect anything?
I checked and I think it is fine, but could use a second pair of eyes.
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/shuffle/KafkaShuffleITCase.java:
##########
@@ -374,17 +371,16 @@ private void testRecordSerDe(TimeCharacteristic timeCharacteristic) throws Excep
// Records in a single partition are kept in order
Collection<ConsumerRecord<byte[], byte[]>> records =
- Iterables.getOnlyElement(
- testKafkaShuffleProducer(
- topic(
- "test_serde-" + UUID.randomUUID(),
- timeCharacteristic),
- env,
- 1,
- 1,
- numElementsPerProducer,
- timeCharacteristic)
- .values());
+ testKafkaShuffleProducer(
+ topic("test_serde-" + UUID.randomUUID(), timeCharacteristic),
+ env,
+ 1,
+ 1,
+ numElementsPerProducer,
+ timeCharacteristic)
+ .values()
+ .iterator()
+ .next();
Review Comment:
Moving away from `Iterables.getOnlyElement`, you'd lose the test coverage where we test there's actually only 1 element in the iterable. Can you add an additional assertion where there is no more next element, to retain that coverage?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317627531
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -245,12 +245,12 @@ public void close() throws Exception {
LOG.debug("Closing writer with {}", currentProducer);
closeAll(
this::abortCurrentProducer,
- closer,
producerPool::clear,
() -> {
checkState(currentProducer.isClosed());
currentProducer = null;
});
+ closeAll(closeables);
Review Comment:
Actually I think there is, because of the `checkState(currentProducer.isClosed())` in the first `closeAll` above.
The `closeables` need to be closed first, for that check to pass. I'm surprised that no unit tests test the `close` method though, that would have failed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] Gerrrr commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "Gerrrr (via GitHub)" <gi...@apache.org>.
Gerrrr commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317650559
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -245,12 +245,12 @@ public void close() throws Exception {
LOG.debug("Closing writer with {}", currentProducer);
closeAll(
this::abortCurrentProducer,
- closer,
producerPool::clear,
() -> {
checkState(currentProducer.isClosed());
currentProducer = null;
});
+ closeAll(closeables);
Review Comment:
Thank you for catching this! Turns out, integration test failures were due to the incorrect ordering here. For example, `KafkaSinkITCase` failed with
```
2023-09-06T06:59:29.8283020Z java.lang.IllegalStateException: null
2023-09-06T06:59:29.8283703Z at org.apache.flink.util.Preconditions.checkState(Preconditions.java:177) ~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8284351Z at org.apache.flink.connector.kafka.sink.KafkaWriter.lambda$close$0(KafkaWriter.java:250) ~[classes/:?]
2023-09-06T06:59:29.8285042Z at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:255) ~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8285660Z at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:234) ~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8286306Z at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:223) ~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8286870Z at org.apache.flink.connector.kafka.sink.KafkaWriter.close(KafkaWriter.java:246) ~[classes/:?]
2023-09-06T06:59:29.8287538Z at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:255) ~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8288146Z at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:234) ~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8288751Z at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:223) ~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8290025Z at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.close(SinkWriterOperator.java:216) ~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8311878Z at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:163) ~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8313214Z at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.closeAllOperators(RegularOperatorChain.java:125) ~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8314283Z at org.apache.flink.streaming.runtime.tasks.StreamTask.closeAllOperators(StreamTask.java:1043) ~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8315237Z at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:911) ~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8316141Z at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:794) ~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8317027Z at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952) ~[flink-runtime-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8317875Z at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931) ~[flink-runtime-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8318834Z at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) ~[flink-runtime-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8319526Z at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) ~[flink-runtime-1.17.0.jar:1.17.0]
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] Gerrrr commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "Gerrrr (via GitHub)" <gi...@apache.org>.
Gerrrr commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317655417
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -104,9 +103,10 @@ class KafkaWriter<IN>
// producer pool only used for exactly once
private final Deque<FlinkKafkaInternalProducer<byte[], byte[]>> producerPool =
new ArrayDeque<>();
- private final Closer closer = Closer.create();
private long lastCheckpointId;
+ private final Set<AutoCloseable> closeables = new HashSet<>();
Review Comment:
I renamed the variable to `producerCloseables` and changed its type to `Deque` in [dfcfdb4](https://github.com/apache/flink-connector-kafka/pull/46/commits/dfcfdb47fabf63632edac94139f9acb3c36eb3b5). I chose `Deque` because this is the type used by [Closer](https://github.com/google/guava/blob/3dc3312423209098720e638ac3cb6884fbe77b75/guava/src/com/google/common/io/Closer.java#L111).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] Gerrrr commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "Gerrrr (via GitHub)" <gi...@apache.org>.
Gerrrr commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1316549706
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -279,7 +279,8 @@ FlinkKafkaInternalProducer<byte[], byte[]> getCurrentProducer() {
void abortLingeringTransactions(
Collection<KafkaWriterState> recoveredStates, long startCheckpointId) {
- List<String> prefixesToAbort = Lists.newArrayList(transactionalIdPrefix);
+ List<String> prefixesToAbort = new ArrayList<>();
Review Comment:
_note_: I did not change this one to `Arrays.asList` because that method returns an immutable view while `prefixesToAbort` is updated a couple of lines below.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] Gerrrr commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "Gerrrr (via GitHub)" <gi...@apache.org>.
Gerrrr commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1316546163
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -452,12 +453,9 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
asyncProducerException = decorateException(metadata, exception, producer);
}
+ // Checking for exceptions from previous writes
mailboxExecutor.submit(
- () -> {
- // Checking for exceptions from previous writes
- checkAsyncException();
- },
- "Update error metric");
+ KafkaWriter.this::checkAsyncException, "Update error metric");
Review Comment:
Fixed in 8362d9a053b6ff41d740135373200c1f3d3daafc
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#issuecomment-1716166328
Closed via 818d1fdedaad63631eab5d44ec90c748cfcf299f
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317636517
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -104,9 +103,10 @@ class KafkaWriter<IN>
// producer pool only used for exactly once
private final Deque<FlinkKafkaInternalProducer<byte[], byte[]>> producerPool =
new ArrayDeque<>();
- private final Closer closer = Closer.create();
private long lastCheckpointId;
+ private final Set<AutoCloseable> closeables = new HashSet<>();
Review Comment:
Also, should this be a `Queue` / `List` instead? We probably never want actual implementation of the producer's equality check to interfere here and leave room to potentially have a producer left unclosed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317636517
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -104,9 +103,10 @@ class KafkaWriter<IN>
// producer pool only used for exactly once
private final Deque<FlinkKafkaInternalProducer<byte[], byte[]>> producerPool =
new ArrayDeque<>();
- private final Closer closer = Closer.create();
private long lastCheckpointId;
+ private final Set<AutoCloseable> closeables = new HashSet<>();
Review Comment:
Also, should this be a `Queue` / `List` instead? We probably never want actual implementation of the producer object's equality check to interfere here and leave room to potentially have a producer left unclosed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317660703
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -245,12 +245,12 @@ public void close() throws Exception {
LOG.debug("Closing writer with {}", currentProducer);
closeAll(
this::abortCurrentProducer,
- closer,
producerPool::clear,
() -> {
checkState(currentProducer.isClosed());
currentProducer = null;
});
+ closeAll(closeables);
Review Comment:
always makes me happy to see tests fail meaningfully :)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] Gerrrr commented on a diff in pull request #46: [FLINK-33017] Remove dependency on shaded guava
Posted by "Gerrrr (via GitHub)" <gi...@apache.org>.
Gerrrr commented on code in PR #46:
URL: https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1316546436
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -245,12 +245,12 @@ public void close() throws Exception {
LOG.debug("Closing writer with {}", currentProducer);
closeAll(
this::abortCurrentProducer,
- closer,
producerPool::clear,
() -> {
checkState(currentProducer.isClosed());
currentProducer = null;
});
+ closeAll(closeables);
Review Comment:
I do not think that the closing order has any correctness implications here. If you prefer, I can keep the original ordering, just in case.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java:
##########
@@ -173,9 +172,10 @@ private static Properties getProperties() {
}
private static List<Consumer<FlinkKafkaInternalProducer<?, ?>>> provideTransactionsFinalizer() {
- return Lists.newArrayList(
- FlinkKafkaInternalProducer::commitTransaction,
- FlinkKafkaInternalProducer::abortTransaction);
+ List<Consumer<FlinkKafkaInternalProducer<?, ?>>> ret = new ArrayList<>();
+ ret.add(FlinkKafkaInternalProducer::commitTransaction);
+ ret.add(FlinkKafkaInternalProducer::abortTransaction);
Review Comment:
Done in 731fc35b2f0204f162044950abcdcfde570f12ef
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org