Refactor functions to use Sink interface #1708
Conversation
|
@sijie @srkukarni please review. |
| CompletableFuture<Void> write(T value) throws Exception; |
There was a problem hiding this comment.
Exception should be thrown through CompletableFuture<Void>, no?
It is a bit strange to throw exception for a method that returns CompletableFuture
There was a problem hiding this comment.
I agree. I will change
| void close() throws Exception; | ||
| } | ||
|
|
||
| private class PulsarSinkAtMostOnceProcessor implements PulsarSinkProcessor { |
There was a problem hiding this comment.
this cannot be static. I am using the PulsarClient and PulsarSinkConfig from the parent class
| } | ||
| } | ||
|
|
||
| private class PulsarSinkAtLeastOnceProcessor implements PulsarSinkProcessor { |
There was a problem hiding this comment.
this cannot be static. I am using the PulsarClient and PulsarSinkConfig from the parent class
|
|
||
| // currently on PulsarRecord | ||
| Producer producer = outputProducer.getProducer(pulsarRecord.getTopicName(), | ||
| Integer.parseInt(pulsarRecord.getPartitionId())); |
There was a problem hiding this comment.
I think we can change MultiConsumerOneOutputTopicProducers to use Map<String, Producer>, rather than converting string back to int
|
|
||
| @Override | ||
| public CompletableFuture<Void> write(T value) throws Exception { | ||
| return null; |
There was a problem hiding this comment.
throw new UnsupportedException(..)?
| } | ||
|
|
||
| @Override | ||
| public void close() throws Exception { |
There was a problem hiding this comment.
I think we need to close the producers here, right?
There was a problem hiding this comment.
good catch will add
| @@ -29,7 +32,9 @@ | |||
| * <p>There is a default implementation provided for wrapping up the user provided {@link Sink}. Pulsar sink | |||
| * should be implemented using this interface to ensure supporting effective-once. | |||
| */ | |||
| public interface RuntimeSink<T> extends Sink<T> { | |||
| //public interface RuntimeSink<T> extends Sink<T> { | |||
No description provided.