#88 | Initial idea to support pagination #90

AkshatJindal1 · 2021-01-05T05:48:47Z

This is not a final version to support pagination #88 . This is my idea. Please let me know if this approach looks good.

castorm

Hi @AkshatJindal1,

In general there are some issues with the extension of existing classes for purposes that violate their responsibilities. That shouldn't be a big deal, as it could be managed by finding the right abstractions, maybe a new abstraction is required to deal with the pagination specifics.

There are also some issues with some low level detail that is managed at a place where high level abstractions are used, again, not a big deal, a matter of moving code around.

Also, I would suggest you focus on a single variation of the problem first, until all abstractions are laid out, otherwise you might be adding too much complexity too soon, difficulting the task. I'm referring to the optionality of appendNextPageUrl.

My main concern, though, has nothing to do with the above. It's about how state is being managed. Have you considered what would happen if the connector is restarted at any point?

The state you are relaying on, like modifiedUrl or the body of the previous response, would be gone.

That's why it's important all relevant information sharing happens through the Offset. That's the only information that's persisted and can be used for recovery.

Would there be a way of moving that relevant state to the offset of the records? I would focus on this problem first, before looking for the right abstractions.

Thanks,
Best regards.

castorm · 2021-01-05T09:42:52Z

...main/java/com/github/castorm/kafka/connect/http/response/spi/KvRecordHttpResponseParser.java

 public interface KvRecordHttpResponseParser extends Configurable {

    List<KvRecord> parse(HttpResponse response);

    default void configure(Map<String, ?> map) {
        // Do nothing
    }
+
+    Optional<String> getNextPageUrl(HttpResponse response);


Same as HttpResponseParser.

castorm · 2021-01-05T09:43:59Z

...ttp/src/main/java/com/github/castorm/kafka/connect/http/response/spi/HttpResponseParser.java

 public interface HttpResponseParser extends Configurable {

    List<SourceRecord> parse(HttpResponse response);

    default void configure(Map<String, ?> map) {
        // Do nothing
    }
+
+    Optional<String> getNextPageUrl(HttpResponse response);


The responsibility of this parser is to translate responses to records, this additional responsibility is not cohesive with it.

In other words, we are breaking single responsibility here. If the new functionality were cohesive with the existing one, maybe we could consider raising the abstraction level of this interface to absorb it, but I don't think it is.

castorm · 2021-01-05T09:51:05Z

...java/com/github/castorm/kafka/connect/http/response/jackson/JacksonResponseRecordParser.java

+            serializer.checkIfNonNull(this.jsonBody, pointer)
+                ? serializer.getObjectAt(this.jsonBody, pointer).asText()
+                : null);


I think this would belong better one level down at the serializer, by providing something like this:

Optional<JsonNode> getObjectAtIfPresent(JsonNode node, JsonPointer pointer)

That would mean you wouldn't need to expose checkIfNonNull, as it would be leaving that responsibility closer to where the decision is made.

castorm · 2021-01-05T09:51:53Z

...java/com/github/castorm/kafka/connect/http/response/jackson/JacksonResponseRecordParser.java

@@ -45,6 +46,10 @@

    private JsonPointer recordsPointer;

+    private Optional<JsonPointer> nextPagePointer;
+
+    private JsonNode jsonBody;


This class is meant to be stateless, we shouldn't be keeping this body around here.

castorm · 2021-01-05T09:53:34Z

...n/java/com/github/castorm/kafka/connect/http/response/jackson/JacksonRecordParserConfig.java

@@ -73,6 +77,7 @@ public static ConfigDef config() {
                .define(ITEM_POINTER, STRING, "/", HIGH, "Item JsonPointer")
                .define(ITEM_KEY_POINTER, STRING, null, HIGH, "Item Key JsonPointers")
                .define(ITEM_TIMESTAMP_POINTER, STRING, null, MEDIUM, "Item Timestamp JsonPointer")
-                .define(ITEM_OFFSET_VALUE_POINTER, STRING, "", MEDIUM, "Item Offset JsonPointers");
+                .define(ITEM_OFFSET_VALUE_POINTER, STRING, "", MEDIUM, "Item Offset JsonPointers")
+            .define(NEXT_PAGE_POINTER, STRING, "/next", LOW, "Pointer for next page");


It's probably not fair to assume something as arbitrary "/next" as a default.

castorm · 2021-01-05T09:57:06Z

...p/src/main/java/com/github/castorm/kafka/connect/http/response/PolicyHttpResponseParser.java

+    @Override
+    public Optional<String> getNextPageUrl(HttpResponse response) {
+        switch (policy.resolve(response)) {
+            case PROCESS:
+                return delegate.getNextPageUrl(response);
+            case SKIP:
+                return Optional.empty();
+            case FAIL:
+            default:
+                throw new IllegalStateException(String.format("Policy failed for response code: %s, body: %s", response.getCode(), ofNullable(response.getBody()).map(String::new).orElse("")));
+        }
+    }


The purpose of the PolicyHttpResponseParser is to enforce a policy for whether response should be parsed, skipped, or failed based on http response status codes.

What's the reasoning to fit this functionality here? isn't it unrelated to the purpose of the class?

castorm · 2021-01-05T10:03:34Z

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/HttpSourceTask.java

+            request = HttpRequest.builder()
+                .method(request.getMethod())
+                .url(modifiedUrl)
+                .headers(request.getHeaders())
+                .body(request.getBody())
+                .build();


We are creating a request here. Why not doing it in the entity meant for that: RequestFactory?

You might want to create a RequestFactory specific for this use case though.

I tried to create a different request factory to take care of pagination. But I am stuck at a point. I am not able to understand how to pass the next page URL from the response to this request factory to create a new request.

castorm · 2021-01-05T10:04:30Z

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/HttpSourceTask.java

+        if(handlePagination) {
+            Optional<String> nextPageUrl = responseParser.getNextPageUrl(response);
+            log.info("Next page URL: {}", nextPageUrl.orElse("no value"));
+            if( isNextPageUrlPresent(nextPageUrl) ) {
+                modifiedUrl = appendNextPageUrl
+                    ? baseUrl + nextPageUrl.orElse("")
+                    : nextPageUrl.orElse(null);
+            } else {
+                modifiedUrl = null;
+            }
+        }


This should be extracted somewhere else, it doesn't seem to be at the same level of abstraction as the rest of the code in this method. Too detailed and specific about pagination.

castorm · 2021-01-05T10:06:37Z

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/HttpSourceTask.java

+
+    private String modifiedUrl;
+
+    private HttpRequest request = null;


Why do you need this here?

AkshatJindal1 mentioned this pull request Jan 5, 2021

Support for paginated response from API #88

Open

castorm#88 | Initial idea to support pagination

28e173f

AkshatJindal1 force-pushed the feature/pagination-support branch from 1c19100 to 28e173f Compare January 5, 2021 05:57

castorm requested changes Jan 5, 2021

View reviewed changes

castorm linked an issue Jan 5, 2021 that may be closed by this pull request

Support for paginated response from API #88

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

#88 | Initial idea to support pagination #90

#88 | Initial idea to support pagination #90

AkshatJindal1 commented Jan 5, 2021

castorm left a comment

castorm Jan 5, 2021

castorm Jan 5, 2021

castorm Jan 5, 2021

castorm Jan 5, 2021

castorm Jan 5, 2021

castorm Jan 5, 2021

castorm Jan 5, 2021

AkshatJindal1 Jan 12, 2021

castorm Jan 5, 2021

castorm Jan 5, 2021


		private String modifiedUrl;

		private HttpRequest request = null;

#88 | Initial idea to support pagination #90

Are you sure you want to change the base?

#88 | Initial idea to support pagination #90

Conversation

AkshatJindal1 commented Jan 5, 2021

castorm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment