Recent updates in Confluent Cloud have improved the reliability of queries. (#11)

This negates the need for the retries that were previously used.
The retry logic has been removed as a consequence.

Author: WadeWaldron

CHANGELOG.md

@@ -23,4 +23,10 @@
 ## Version 0.3.3
 
 * Added a dev container file.
-* Updated the Gitpod configuration.
+* Updated the Gitpod configuration.
+
+## Version 0.3.4
+
+* Recent updates in Confluent Cloud have improved the reliability of queries. 
+This negates the need for the retries that were previously used.
+The retry logic has been removed as a consequence.

solutions/01-connecting-to-confluent-cloud/pom.xml

@@ -198,7 +198,6 @@ under the License.
                     </dependency>
                 </dependencies>
                 <configuration>
-                    <rerunFailingTestsCount>5</rerunFailingTestsCount>
                     <argLine>-XX:+EnableDynamicAgentLoading</argLine>
                 </configuration>
             </plugin>

solutions/02-querying-flink-tables/pom.xml

@@ -198,7 +198,6 @@ under the License.
                     </dependency>
                 </dependencies>
                 <configuration>
-                    <rerunFailingTestsCount>5</rerunFailingTestsCount>
                     <argLine>-XX:+EnableDynamicAgentLoading</argLine>
                 </configuration>
             </plugin>

solutions/02-querying-flink-tables/src/test/java/marketplace/CustomerServiceIntegrationTest.java

@@ -56,7 +56,7 @@ public void allCustomers_shouldReturnTheDetailsOfAllCustomers() throws Exception
         env.fromValues(customers).insertInto(customersTableName).execute();
 
         // Execute the query.
-        TableResult results = retry(() -> customerService.allCustomers());
+        TableResult results = customerService.allCustomers();
 
         // Fetch the actual results.
         List<Row> actual = fetchRows(results)
@@ -90,7 +90,7 @@ public void allCustomerAddresses_shouldReturnTheAddressesOfAllCustomers() throws
         env.fromValues(customers).insertInto(customersTableName).execute();
 
         // Execute the query.
-        TableResult results = retry(() -> customerService.allCustomerAddresses());
+        TableResult results = customerService.allCustomerAddresses();
 
         // Fetch the actual results.
         List<Row> actual = fetchRows(results)

solutions/02-querying-flink-tables/src/test/java/marketplace/FlinkIntegrationTest.java

@@ -93,26 +93,6 @@ public void mainTeardown() {
         }
     }
 
-    protected TableResult retry(Supplier<TableResult> supplier) {
-        return retry(3, supplier);
-    }
-
-    protected TableResult retry(int tries, Supplier<TableResult> supplier) {
-        try {
-            return supplier.get();
-        } catch (Exception e) {
-            logger.error("Failed on retryable command.", e);
-
-            if(tries > 0) {
-                logger.info("Retrying");
-                return retry(tries - 1, supplier);
-            } else {
-                logger.info("Maximum number of tries exceeded. Failing...");
-                throw e;
-            }
-        }
-    }
-
     protected TableResult cancelOnExit(TableResult tableResult) {
         jobsToCancel.add(tableResult);
         return tableResult;

solutions/02-querying-flink-tables/src/test/java/marketplace/OrderServiceIntegrationTest.java

@@ -72,7 +72,7 @@ public void ordersOver50Dollars_shouldOnlyReturnOrdersWithAPriceOf50DollarsOrMor
         env.fromValues(orders).insertInto(ordersTableName).execute();
 
         // Execute the query.
-        TableResult results = retry(() -> orderService.ordersOver50Dollars());
+        TableResult results = orderService.ordersOver50Dollars();
 
         // Build the expected results.
         List<Row> expected = orders.stream().filter(row -> row.<Double>getFieldAs(indexOf("price")) >= 50).toList();
@@ -114,7 +114,7 @@ public void pricesWithTax_shouldReturnTheCorrectPrices() {
         env.fromValues(orders).insertInto(ordersTableName).execute();
 
         // Execute the query.
-        TableResult results = retry(() -> orderService.pricesWithTax(taxAmount));
+        TableResult results = orderService.pricesWithTax(taxAmount);
 
         // Fetch the actual results.
         List<Row> actual = fetchRows(results)

solutions/03-building-a-streaming-pipeline/pom.xml

@@ -198,7 +198,6 @@ under the License.
                     </dependency>
                 </dependencies>
                 <configuration>
-                    <rerunFailingTestsCount>5</rerunFailingTestsCount>
                     <argLine>-XX:+EnableDynamicAgentLoading</argLine>
                 </configuration>
             </plugin>

solutions/03-building-a-streaming-pipeline/src/test/java/marketplace/CustomerServiceIntegrationTest.java

@@ -56,7 +56,7 @@ public void allCustomers_shouldReturnTheDetailsOfAllCustomers() throws Exception
         env.fromValues(customers).insertInto(customersTableName).execute();
 
         // Execute the query.
-        TableResult results = retry(() -> customerService.allCustomers());
+        TableResult results = customerService.allCustomers();
 
         // Fetch the actual results.
         List<Row> actual = fetchRows(results)
@@ -90,7 +90,7 @@ public void allCustomerAddresses_shouldReturnTheAddressesOfAllCustomers() throws
         env.fromValues(customers).insertInto(customersTableName).execute();
 
         // Execute the query.
-        TableResult results = retry(() -> customerService.allCustomerAddresses());
+        TableResult results = customerService.allCustomerAddresses();
 
         // Fetch the actual results.
         List<Row> actual = fetchRows(results)

solutions/03-building-a-streaming-pipeline/src/test/java/marketplace/FlinkIntegrationTest.java

@@ -93,26 +93,6 @@ public void mainTeardown() {
         }
     }
 
-    protected TableResult retry(Supplier<TableResult> supplier) {
-        return retry(3, supplier);
-    }
-
-    protected TableResult retry(int tries, Supplier<TableResult> supplier) {
-        try {
-            return supplier.get();
-        } catch (Exception e) {
-            logger.error("Failed on retryable command.", e);
-
-            if(tries > 0) {
-                logger.info("Retrying");
-                return retry(tries - 1, supplier);
-            } else {
-                logger.info("Maximum number of tries exceeded. Failing...");
-                throw e;
-            }
-        }
-    }
-
     protected TableResult cancelOnExit(TableResult tableResult) {
         jobsToCancel.add(tableResult);
         return tableResult;

solutions/03-building-a-streaming-pipeline/src/test/java/marketplace/OrderServiceIntegrationTest.java

@@ -86,7 +86,7 @@ public void ordersOver50Dollars_shouldOnlyReturnOrdersWithAPriceOf50DollarsOrMor
         env.fromValues(orders).insertInto(ordersTableName).execute();
 
         // Execute the query.
-        TableResult results = retry(() -> orderService.ordersOver50Dollars());
+        TableResult results = orderService.ordersOver50Dollars();
 
         // Build the expected results.
         List<Row> expected = orders.stream().filter(row -> row.<Double>getFieldAs(indexOf("price")) >= 50).toList();
@@ -128,7 +128,7 @@ public void pricesWithTax_shouldReturnTheCorrectPrices() {
         env.fromValues(orders).insertInto(ordersTableName).execute();
 
         // Execute the query.
-        TableResult results = retry(() -> orderService.pricesWithTax(taxAmount));
+        TableResult results = orderService.pricesWithTax(taxAmount);
 
         // Fetch the actual results.
         List<Row> actual = fetchRows(results)
@@ -162,6 +162,7 @@ public void pricesWithTax_shouldReturnTheCorrectPrices() {
     @Test
     @Timeout(60)
     public void createFreeShippingTable_shouldCreateTheTable() {
+        deleteTable(orderQualifiedForFreeShippingTableName);
         deleteTableOnExit(orderQualifiedForFreeShippingTableName);
 
         TableResult result = orderService.createFreeShippingTable();
@@ -221,11 +222,9 @@ public void streamOrdersOver50Dollars_shouldStreamRecordsToTheTable() throws Exc
         cancelOnExit(orderService.streamOrdersOver50Dollars());
 
         // Query the destination table.
-        TableResult queryResult = retry(() ->
-            env.from(orderQualifiedForFreeShippingTableName)
-                .select($("*"))
-                .execute()
-        );
+        TableResult queryResult = env.from(orderQualifiedForFreeShippingTableName)
+            .select($("*"))
+            .execute();
 
         // Obtain the actual results.
         List<Row> actual = fetchRows(queryResult)